Prep for #2179. Abstract parsing of frame header from Session into struct. Catch unhandled exceptions thrown by ASIO.

libp2p/Common.h

@@ -149,14 +149,15 @@ using CapDescs = std::vector<CapDesc>;
  */
 struct PeerSessionInfo
 {
-	NodeId id;
-	std::string clientVersion;
-	std::string host;
-	unsigned short port;
+	NodeId const id;
+	std::string const clientVersion;
+	std::string const host;
+	unsigned short const port;
 	std::chrono::steady_clock::duration lastPing;
-	std::set<CapDesc> caps;
+	std::set<CapDesc> const caps;
 	unsigned socketId;
 	std::map<std::string, std::string> notes;
+	unsigned const protocolVersion;
 };
 
 using PeerSessionInfos = std::vector<PeerSessionInfo>;

libp2p/Host.cpp

@@ -254,7 +254,7 @@ void Host::startPeerSession(Public const& _id, RLP const& _rlp, RLPXFrameCoder*
 	clog(NetMessageSummary) << "Hello: " << clientVersion << "V[" << protocolVersion << "]" << _id << showbase << capslog.str() << dec << listenPort;
 	
 	// create session so disconnects are managed
-	auto ps = make_shared<Session>(this, _io, _s, p, PeerSessionInfo({_id, clientVersion, p->endpoint.address.to_string(), listenPort, chrono::steady_clock::duration(), _rlp[2].toSet<CapDesc>(), 0, map<string, string>()}));
+	auto ps = make_shared<Session>(this, _io, _s, p, PeerSessionInfo({_id, clientVersion, p->endpoint.address.to_string(), listenPort, chrono::steady_clock::duration(), _rlp[2].toSet<CapDesc>(), 0, map<string, string>(), protocolVersion}));
 	if (protocolVersion < dev::p2p::c_protocolVersion - 1)
 	{
 		ps->disconnect(IncompatibleProtocol);
@@ -723,10 +723,18 @@ void Host::startedWorking()
 }
 
 void Host::doWork()
+{
+	try
 	{
 		if (m_run)
 			m_ioService.run();
 	}
+	catch (std::exception const& _e)
+	{
+		clog(NetP2PWarn) << "Exception in Network Thread:" << _e.what();
+		clog(NetP2PWarn) << "Network Restart is Recommended.";
+	}
+}
 
 void Host::keepAlivePeers()
 {

libp2p/RLPXFrameCoder.cpp

@@ -29,6 +29,18 @@ using namespace dev;
 using namespace dev::p2p;
 using namespace CryptoPP;
 
+RLPXFrameInfo::RLPXFrameInfo(bytesConstRef _header)
+{
+	length = (_header[0] * 256 + _header[1]) * 256 + _header[2];
+	padding = ((16 - (length % 16)) % 16);
+	RLP header(_header.cropped(3), RLP::ThrowOnFail | RLP::FailIfTooSmall);
+	auto itemCount = header.itemCount();
+	protocolId = header[0].toInt<uint16_t>();
+	hasSequence = itemCount > 1;
+	sequenceId = hasSequence ? header[1].toInt<uint16_t>() : 0;
+	totalLength = itemCount == 3 ? header[2].toInt<uint32_t>() : 0;
+}
+
 RLPXFrameCoder::RLPXFrameCoder(RLPXHandshake const& _init)
 {
 	// we need:

libp2p/RLPXFrameCoder.h

@@ -33,6 +33,20 @@ namespace dev
 namespace p2p
 {
 
+struct RLPXFrameInfo
+{
+	RLPXFrameInfo() = default;
+	/// Constructor. frame-size || protocol-type, [sequence-id[, total-packet-size]]
+	RLPXFrameInfo(bytesConstRef _frameHeader);
+	uint32_t length = 0;			///< Max: 2**24
+	uint8_t padding = 0;
+	
+	uint16_t protocolId = 0;
+	bool hasSequence = false;
+	uint16_t sequenceId = 0;
+	uint32_t totalLength = 0;
+};
+
 class RLPXHandshake;
 
 /**

libp2p/Session.cpp

@@ -27,7 +27,6 @@
 #include <libdevcore/CommonIO.h>
 #include <libdevcore/StructuredLogger.h>
 #include <libethcore/Exceptions.h>
-#include "RLPxHandshake.h"
 #include "Host.h"
 #include "Capability.h"
 using namespace std;
@@ -156,12 +155,36 @@ void Session::serviceNodesRequest()
 	addNote("peers", "done");
 }
 
-bool Session::interpret(PacketType _t, RLP const& _r)
+bool Session::frameReceived(uint16_t _capId, PacketType _t, RLP const& _r)
 {
 	m_lastReceived = chrono::steady_clock::now();
-
 	clog(NetRight) << _t << _r;
 	try		// Generic try-catch block designed to capture RLP format errors - TODO: give decent diagnostics, make a bit more specific over what is caught.
+	{
+		// v4 frame headers are useless, offset packet type used
+		// v5 protocol type is in header, packet type not offset
+		if (_capId == 0 && _t < UserPacket)
+			return interpret(_t, _r);
+		if (m_info.protocolVersion >= 5)
+			for (auto const& i: m_capabilities)
+				if (_capId == (uint16_t)i.first.second)
+					return i.second->m_enabled ? i.second->interpret(_t, _r) : true;
+		if (m_info.protocolVersion <= 4)
+			for (auto const& i: m_capabilities)
+				if (_t >= (int)i.second->m_idOffset && _t - i.second->m_idOffset < i.second->hostCapability()->messageCount())
+					return i.second->m_enabled ? i.second->interpret(_t - i.second->m_idOffset, _r) : true;
+		return false;
+	}
+	catch (std::exception const& _e)
+	{
+		clog(NetWarn) << "Peer causing an exception:" << _e.what() << _r;
+		disconnect(BadProtocol);
+		return true;
+	}
+	return true;
+}
+
+bool Session::interpret(PacketType _t, RLP const& _r)
 {
 	switch (_t)
 	{
@@ -251,23 +274,8 @@ bool Session::interpret(PacketType _t, RLP const& _r)
 		}
 		break;
 	default:
-			for (auto const& i: m_capabilities)
-				if (_t >= (int)i.second->m_idOffset && _t - i.second->m_idOffset < i.second->hostCapability()->messageCount())
-				{
-					if (i.second->m_enabled)
-						return i.second->interpret(_t - i.second->m_idOffset, _r);
-					else
-						return true;
-				}
 		return false;
 	}
-	}
-	catch (std::exception const& _e)
-	{
-		clog(NetWarn) << "Peer causing an exception:" << _e.what() << _r;
-		disconnect(BadProtocol);
-		return true;
-	}
 	return true;
 }
 
@@ -292,12 +300,9 @@ void Session::sealAndSend(RLPStream& _s)
 
 bool Session::checkPacket(bytesConstRef _msg)
 {
-	if (_msg.size() < 2)
+	if (_msg[0] > 0x7f || _msg.size() < 2)
 		return false;
-	if (_msg[0] > 0x7f)
-		return false;
-	RLP r(_msg.cropped(1));
-	if (r.actualSize() + 1 != _msg.size())
+	if (RLP(_msg.cropped(1)).actualSize() + 1 != _msg.size())
 		return false;
 	return true;
 }
@@ -413,65 +418,43 @@ void Session::doRead()
 	{
 		ThreadContext tc(info().id.abridged());
 		ThreadContext tc2(info().clientVersion);
-		if (ec && ec.category() != boost::asio::error::get_misc_category() && ec.value() != boost::asio::error::eof)
-		{
-			clog(NetWarn) << "Error reading: " << ec.message();
-			drop(TCPError);
-		}
-		else if (ec && length == 0)
+		if (!checkRead(h256::size, ec, length))
 			return;
-		else
-		{
-			/// authenticate and decrypt header
-			bytesRef header(m_data.data(), h256::size);
-			if (!m_io->authAndDecryptHeader(header))
+		else if (!m_io->authAndDecryptHeader(bytesRef(m_data.data(), length)))
 		{
 			clog(NetWarn) << "header decrypt failed";
 			drop(BadProtocol); // todo: better error
 			return;
 		}
 
-			/// check frame size
-			uint32_t frameSize = (m_data[0] * 256 + m_data[1]) * 256 + m_data[2];
-			if (frameSize >= (uint32_t)1 << 24)
+		RLPXFrameInfo header;
+		try
 		{
-				clog(NetWarn) << "frame size too large";
+			header = RLPXFrameInfo(bytesConstRef(m_data.data(), length));
+		}
+		catch (std::exception const& _e)
+		{
+			clog(NetWarn) << "Exception decoding frame header RLP:" << bytesConstRef(m_data.data(), h128::size).cropped(3);
 			drop(BadProtocol);
 			return;
 		}
 		
-			/// rlp of header has protocol-type, sequence-id[, total-packet-size]
-			bytes headerRLP(13);
-			bytesConstRef(m_data.data(), h128::size).cropped(3).copyTo(&headerRLP);
-			
 		/// read padded frame and mac
-			auto tlen = frameSize + ((16 - (frameSize % 16)) % 16) + h128::size;
-			ba::async_read(m_socket->ref(), boost::asio::buffer(m_data, tlen), [this, self, headerRLP, frameSize, tlen](boost::system::error_code ec, std::size_t length)
+		auto tlen = header.length + header.padding + h128::size;
+		ba::async_read(m_socket->ref(), boost::asio::buffer(m_data, tlen), [this, self, header, tlen](boost::system::error_code ec, std::size_t length)
 		{
 			ThreadContext tc(info().id.abridged());
 			ThreadContext tc2(info().clientVersion);
-				if (ec && ec.category() != boost::asio::error::get_misc_category() && ec.value() != boost::asio::error::eof)
-				{
-					clog(NetWarn) << "Error reading: " << ec.message();
-					drop(TCPError);
-				}
-				else if (ec && length < tlen)
-				{
-					clog(NetWarn) << "Error reading - Abrupt peer disconnect: " << ec.message();
-					repMan().noteRude(*this);
-					drop(TCPError);
+			if (!checkRead(tlen, ec, length))
 				return;
-				}
-				else
-				{
-					if (!m_io->authAndDecryptFrame(bytesRef(m_data.data(), tlen)))
+			else if (!m_io->authAndDecryptFrame(bytesRef(m_data.data(), tlen)))
 			{
 				clog(NetWarn) << "frame decrypt failed";
 				drop(BadProtocol); // todo: better error
 				return;
 			}
 
-					bytesConstRef frame(m_data.data(), frameSize);
+			bytesConstRef frame(m_data.data(), header.length);
 			if (!checkPacket(frame))
 			{
 				cerr << "Received " << frame.size() << ": " << toHex(frame) << endl;
@@ -483,12 +466,30 @@ void Session::doRead()
 			{
 				auto packetType = (PacketType)RLP(frame.cropped(0, 1)).toInt<unsigned>();
 				RLP r(frame.cropped(1));
-						if (!interpret(packetType, r))
+				if (!frameReceived(header.protocolId, packetType, r))
 					clog(NetWarn) << "Couldn't interpret packet." << RLP(r);
 			}
 			doRead();
-				}
 		});
-		}
 	});
 }
+
+bool Session::checkRead(std::size_t _expected, boost::system::error_code _ec, std::size_t _length)
+{
+	if (_ec && _ec.category() != boost::asio::error::get_misc_category() && _ec.value() != boost::asio::error::eof)
+	{
+		clog(NetConnect) << "Error reading: " << _ec.message();
+		drop(TCPError);
+		return false;
+	}
+	else if (_ec && _length < _expected)
+	{
+		clog(NetWarn) << "Error reading - Abrupt peer disconnect: " << _ec.message();
+		repMan().noteRude(*this);
+		drop(TCPError);
+		return false;
+	}
+	// If this fails then there's an unhandled asio error
+	assert(_expected == _length);
+	return true;
+}

libp2p/Session.h

@@ -97,10 +97,16 @@ private:
 	/// Perform a read on the socket.
 	void doRead();
 	
+	/// Check error code after reading and drop peer if error code.
+	bool checkRead(std::size_t _expected, boost::system::error_code _ec, std::size_t _length);
+
 	/// Perform a single round of the write operation. This could end up calling itself asynchronously.
 	void write();
 
-	/// Interpret an incoming message.
+	/// Deliver RLPX packet to Session or Capability for interpretation.
+	bool frameReceived(uint16_t _capId, PacketType _t, RLP const& _r);
+
+	/// Interpret an incoming Session packet.
 	bool interpret(PacketType _t, RLP const& _r);
 	
 	/// @returns true iff the _msg forms a valid message for sending or receiving on the network.

test/libp2p/net.cpp

@@ -314,23 +314,23 @@ BOOST_AUTO_TEST_CASE(kademlia)
 	node.nodeTable->discover(); // ideally, joining with empty node table logs warning we can check for
 	node.setup();
 	node.populate();
-	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
+//	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
 
 	node.populateAll();
-	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
+//	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
 
 	auto nodes = node.nodeTable->nodes();
 	nodes.sort();
 
 	node.nodeTable->reset();
-	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
+//	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
 
 	node.populate(1);
-	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
+//	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
 
 	node.nodeTable->discover();
 	this_thread::sleep_for(chrono::milliseconds(2000));
-	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
+//	clog << "NodeTable:\n" << *node.nodeTable.get() << endl;
 
 	BOOST_REQUIRE_EQUAL(node.nodeTable->count(), 8);