Add test for frame writer. Cleanup abstraction in preparation for reader. Initial reader code.

libp2p/RLPXFrameCoder.cpp

@@ -32,22 +32,26 @@ using namespace CryptoPP;
 
 RLPXFrameCoder::RLPXFrameCoder(RLPXHandshake const& _init)
 {
-	// we need:
-	// originated?
-	// Secret == output of ecdhe agreement
-	// authCipher
-	// ackCipher
+	setup(_init.m_originated, _init.m_remoteEphemeral, _init.m_remoteNonce, _init.m_ecdhe, _init.m_nonce, &_init.m_ackCipher, &_init.m_authCipher);
+}
+
+RLPXFrameCoder::RLPXFrameCoder(bool _originated, h512 _remoteEphemeral, h256 _remoteNonce, crypto::ECDHE const& _ecdhe, h256 _nonce, bytesConstRef _ackCipher, bytesConstRef _authCipher)
+{
+	setup(_originated, _remoteEphemeral, _remoteNonce, _ecdhe, _nonce, _ackCipher, _authCipher);
+}
 
+void RLPXFrameCoder::setup(bool _originated, h512 _remoteEphemeral, h256 _remoteNonce, crypto::ECDHE const& _ecdhe, h256 _nonce, bytesConstRef _ackCipher, bytesConstRef _authCipher)
+{
 	bytes keyMaterialBytes(64);
 	bytesRef keyMaterial(&keyMaterialBytes);
-
+	
 	// shared-secret = sha3(ecdhe-shared-secret || sha3(nonce || initiator-nonce))
 	Secret ephemeralShared;
-	_init.m_ecdhe.agree(_init.m_remoteEphemeral, ephemeralShared);
+	_ecdhe.agree(_remoteEphemeral, ephemeralShared);
 	ephemeralShared.ref().copyTo(keyMaterial.cropped(0, h256::size));
 	h512 nonceMaterial;
-	h256 const& leftNonce = _init.m_originated ? _init.m_remoteNonce : _init.m_nonce;
-	h256 const& rightNonce = _init.m_originated ? _init.m_nonce : _init.m_remoteNonce;
+	h256 const& leftNonce = _originated ? _remoteNonce : _nonce;
+	h256 const& rightNonce = _originated ? _nonce : _remoteNonce;
 	leftNonce.ref().copyTo(nonceMaterial.ref().cropped(0, h256::size));
 	rightNonce.ref().copyTo(nonceMaterial.ref().cropped(h256::size, h256::size));
 	auto outRef(keyMaterial.cropped(h256::size, h256::size));
@@ -65,31 +69,31 @@ RLPXFrameCoder::RLPXFrameCoder(RLPXHandshake const& _init)
 	h128 iv;
 	m_frameEnc.SetKeyWithIV(m_frameEncKey, h256::size, iv.data());
 	m_frameDec.SetKeyWithIV(m_frameDecKey, h256::size, iv.data());
-
+	
 	// mac-secret = sha3(ecdhe-shared-secret || aes-secret)
 	sha3(keyMaterial, outRef); // output mac-secret
 	m_macEncKey.resize(h256::size);
 	memcpy(m_macEncKey.data(), outRef.data(), h256::size);
 	m_macEnc.SetKey(m_macEncKey, h256::size);
-
+	
 	// Initiator egress-mac: sha3(mac-secret^recipient-nonce || auth-sent-init)
 	//           ingress-mac: sha3(mac-secret^initiator-nonce || auth-recvd-ack)
 	// Recipient egress-mac: sha3(mac-secret^initiator-nonce || auth-sent-ack)
 	//           ingress-mac: sha3(mac-secret^recipient-nonce || auth-recvd-init)
  
-	(*(h256*)outRef.data() ^ _init.m_remoteNonce).ref().copyTo(keyMaterial);
-	bytes const& egressCipher = _init.m_originated ? _init.m_authCipher : _init.m_ackCipher;
+	(*(h256*)outRef.data() ^ _remoteNonce).ref().copyTo(keyMaterial);
+	bytesConstRef egressCipher = _originated ? _authCipher : _ackCipher;
 	keyMaterialBytes.resize(h256::size + egressCipher.size());
 	keyMaterial.retarget(keyMaterialBytes.data(), keyMaterialBytes.size());
-	bytesConstRef(&egressCipher).copyTo(keyMaterial.cropped(h256::size, egressCipher.size()));
+	egressCipher.copyTo(keyMaterial.cropped(h256::size, egressCipher.size()));
 	m_egressMac.Update(keyMaterial.data(), keyMaterial.size());
-
+	
 	// recover mac-secret by re-xoring remoteNonce
-	(*(h256*)keyMaterial.data() ^ _init.m_remoteNonce ^ _init.m_nonce).ref().copyTo(keyMaterial);
-	bytes const& ingressCipher = _init.m_originated ? _init.m_ackCipher : _init.m_authCipher;
+	(*(h256*)keyMaterial.data() ^ _remoteNonce ^ _nonce).ref().copyTo(keyMaterial);
+	bytesConstRef ingressCipher = _originated ? _ackCipher : _authCipher;
 	keyMaterialBytes.resize(h256::size + ingressCipher.size());
 	keyMaterial.retarget(keyMaterialBytes.data(), keyMaterialBytes.size());
-	bytesConstRef(&ingressCipher).copyTo(keyMaterial.cropped(h256::size, ingressCipher.size()));
+	ingressCipher.copyTo(keyMaterial.cropped(h256::size, ingressCipher.size()));
 	m_ingressMac.Update(keyMaterial.data(), keyMaterial.size());
 }
 

libp2p/RLPXFrameCoder.h

@@ -47,11 +47,17 @@ class RLPXFrameCoder
 	friend class RLPXFrameIOMux;
 	friend class Session;
 public:
-	/// Constructor.
-	/// Requires instance of RLPXHandshake which has completed first two phases of handshake.
+	/// Construct; requires instance of RLPXHandshake which has encrypted ECDH key exchange (first two phases of handshake).
 	RLPXFrameCoder(RLPXHandshake const& _init);
+	
+	/// Construct with external key material.
+	RLPXFrameCoder(bool _originated, h512 _remoteEphemeral, h256 _remoteNonce, crypto::ECDHE const& _ephemeral, h256 _nonce, bytesConstRef _ackCipher, bytesConstRef _authCipher);
+	
 	~RLPXFrameCoder() {}
 	
+	/// Establish shared secrets and setup AES and MAC states. Used by both constructors.
+	void setup(bool _originated, h512 _remoteEphemeral, h256 _remoteNonce, crypto::ECDHE const& _ephemeral, h256 _nonce, bytesConstRef _ackCipher, bytesConstRef _authCipher);
+	
 	/// Write single-frame payload of packet(s).
 	void writeFrame(uint16_t _protocolType, bytesConstRef _payload, bytes& o_bytes);
 

libp2p/RLPXFrameWriter.cpp

@@ -35,7 +35,7 @@ void RLPXFrameWriter::enque(unsigned _packetType, RLPStream& _payload, PacketPri
 size_t RLPXFrameWriter::drain(RLPXFrameCoder& _coder, unsigned _size, vector<bytes>& o_toWrite)
 {
 	static const size_t c_blockSize = h128::size;
-	static const size_t c_overhead = c_blockSize * 2;
+	static const size_t c_overhead = c_blockSize * 3; // header + headerMac + frameMAC
 	if (_size < c_overhead + c_blockSize)
 		return 0;
 
@@ -60,6 +60,7 @@ size_t RLPXFrameWriter::drain(RLPXFrameCoder& _coder, unsigned _size, vector<byt
 		QueueState &qs = high ? m_q.first : m_q.second;
 		size_t frameAllot = (!swapQueues && highPending && lowPending ? frameLen / 2 - (c_overhead + c_blockSize) > 0 ? frameLen / 2 : frameLen : frameLen) - c_overhead;
 		size_t offset = 0;
+		size_t length = 0;
 		while (frameAllot >= c_blockSize)
 		{
 			if (qs.writing == nullptr)
@@ -89,13 +90,13 @@ size_t RLPXFrameWriter::drain(RLPXFrameCoder& _coder, unsigned _size, vector<byt
 			if (frameAllot && qs.remaining)
 			{
 				offset = qs.writing->size() - qs.remaining;
-				auto length = qs.remaining <= frameAllot ? qs.remaining : frameAllot;
+				length = qs.remaining <= frameAllot ? qs.remaining : frameAllot;
 				bytes portion = bytesConstRef(&qs.writing->data()).cropped(offset, length).toBytes();
 				qs.remaining -= length;
 				frameAllot -= portion.size();
 				payload += portion;
 			}
-			if (!qs.remaining)
+			if (!qs.remaining && ret++)
 				qs.writing = nullptr;
 			if (qs.sequenced)
 				break;
@@ -110,6 +111,7 @@ size_t RLPXFrameWriter::drain(RLPXFrameCoder& _coder, unsigned _size, vector<byt
 					_coder.writeFrame(m_protocolType, qs.sequence, &payload, payload);
 			else
 				_coder.writeFrame(m_protocolType, &payload, payload);
+			assert((int)frameLen - payload.size() >= 0);
 			frameLen -= payload.size();
 			o_toWrite.push_back(move(payload));
 			payload.resize(0);
@@ -120,7 +122,6 @@ size_t RLPXFrameWriter::drain(RLPXFrameCoder& _coder, unsigned _size, vector<byt
 				qs.sequenced = false;
 				DEV_GUARDED(qs.x)
 					qs.q.pop_front();
-				ret++;
 			}
 		}
 		else if (swapQueues)

libp2p/RLPXFrameWriter.h

@@ -33,6 +33,63 @@ namespace dev
 namespace p2p
 {
 
+class RLPXFrameReader
+{
+	RLPXFrameReader(uint16_t _protocolType): m_protocolType(_protocolType) {}
+	
+	std::vector<RLPXPacket> assemble(bytes& _in)
+	{
+		bytesConstRef buffer(&_in);
+		std::vector<RLPXPacket> ret;
+		if (!_in.size())
+			return ret;
+		
+		while (!buffer.empty())
+		{
+			auto type = RLPXPacket::nextRLP(buffer);
+			buffer = buffer.cropped(type.size());
+			auto packet = RLPXPacket::nextRLP(buffer);
+			buffer = buffer.cropped(packet.size());
+			RLPXPacket p(m_protocolType, type);
+			p.streamIn(packet);
+			ret.push_back(std::move(p));
+		}
+		return ret;
+	}
+	
+	std::vector<RLPXPacket> assemble(bytes& _in, uint16_t _seq, uint32_t _totalSize)
+	{
+		// TODO: cleanup sequencing api (throw exception(s) when bad things happen)
+		
+		bytesConstRef buffer(&_in);
+		if (!m_incomplete.count(_seq) && _totalSize > 0)
+		{
+			// new packet
+			RLPXPacket p(m_protocolType, buffer);
+			if (p.isValid())
+				return std::vector<RLPXPacket>{std::move(p)};
+			m_incomplete[_seq] = std::move(p);
+		}
+		else
+		{
+			RLPXPacket& p = m_incomplete[_seq];
+			p.streamIn(buffer);
+			if (p.isValid())
+			{
+				std::vector<RLPXPacket> ret{std::move(p)};
+				m_incomplete.erase(_seq);
+				return ret;
+			}
+		}
+
+		return std::vector<RLPXPacket>();
+	}
+	
+protected:
+	uint16_t m_protocolType;
+	std::map<uint16_t, RLPXPacket> m_incomplete;	///< Incomplete packets which span multiple frames.
+};
+
 class RLPXFrameWriter
 {
 	struct QueueState
@@ -47,13 +104,15 @@ class RLPXFrameWriter
 	
 public:
 	enum PacketPriority { PriorityLow = 0, PriorityHigh };
+	static const uint16_t EmptyFrameLength = h128::size * 3; // header + headerMAC + frameMAC
+	static const uint16_t MinFrameDequeLength = h128::size * 4; // header + headerMAC + padded-block + frameMAC
 
 	RLPXFrameWriter(uint16_t _protocolType): m_protocolType(_protocolType) {}
 	RLPXFrameWriter(RLPXFrameWriter const& _s): m_protocolType(_s.m_protocolType) {}
 	
 	size_t size() const { size_t l; size_t h; DEV_GUARDED(m_q.first.x) h = m_q.first.q.size(); DEV_GUARDED(m_q.second.x) l = m_q.second.q.size(); return l + h; }
 	
-	/// Thread-safe.
+	/// Contents of _payload will be moved. Thread-safe.
 	void enque(unsigned _packetType, RLPStream& _payload, PacketPriority _priority = PriorityLow);
 	
 	/// Returns number of packets framed and outputs frames to o_bytes. Not thread-safe.

libp2p/RLPXPacket.h

@@ -28,18 +28,35 @@ namespace dev
 namespace p2p
 {
 
-struct RLPXPacketNullPacket: virtual dev::Exception {};
-	
+struct RLPXNullPacket: virtual dev::Exception {};
+struct RLPXInvalidPacket: virtual dev::Exception {};
+
+/**
+ * RLPX Packet
+ */
 class RLPXPacket
 {
 public:
-	RLPXPacket(unsigned _capId, unsigned _type, RLPStream& _rlps): m_cap(_capId), m_type(_type), m_data(std::move(_rlps.out())) { if (!_type && !m_data.size()) BOOST_THROW_EXCEPTION(RLPXPacketNullPacket()); }
+	static bytesConstRef nextRLP(bytesConstRef _b) { try { RLP r(_b, RLP::AllowNonCanon); auto s = r.actualSize(); if (s >= _b.size()) return _b.cropped(s); } catch(...) {} return bytesConstRef(); }
+	
+	/// Construct complete packet. RLPStream data is moved.
+	RLPXPacket(unsigned _capId, unsigned _type, RLPStream&& _rlps): m_cap(_capId), m_type(_type), m_data(std::move(_rlps.out())) { if (!_type && !m_data.size()) BOOST_THROW_EXCEPTION(RLPXNullPacket()); }
+
+	/// Construct packet with type and initial bytes; type is determined by RLP of 1st byte and packet may be incomplete.
+	RLPXPacket(unsigned _capId, bytesConstRef _in): m_cap(_capId), m_type(getType(_in)) { assert(_in.size()); if (_in.size() > 1) { m_data.resize(_in.size() - 1); _in.cropped(1).copyTo(&m_data); } }
 	
 	unsigned type() const { return m_type; }
 	bytes const& data() const { return m_data; }
 	size_t size() const { return m_data.size(); }
 	
+	bool streamIn(bytesConstRef _in) { auto offset = m_data.size(); m_data.resize(offset + _in.size()); _in.copyTo(byestConstRef(m_data).cropped(offset)); return isValid(); }
+	
+	bool isValid() { if (m_type > 0x7f) return false; try { if (RLP(m_data).actualSize() == m_data.size()) return true; } catch (...) {} return false; }
+	
+
 protected:
+	unsigned getType(bytesConstRef _rlp) { return RLP(_rlp.cropped(1)).toInt<unsigned>(); }
+	
 	unsigned m_cap;
 	unsigned m_type;
 	bytes m_data;

test/libp2p/rlpx.cpp

@@ -31,10 +31,12 @@
 #include <libdevcrypto/ECDHE.h>
 #include <libdevcrypto/CryptoPP.h>
 #include <libp2p/RLPxHandshake.h>
+#include <libp2p/RLPXFrameWriter.h>
 
 using namespace std;
 using namespace dev;
 using namespace dev::crypto;
+using namespace dev::p2p;
 using namespace CryptoPP;
 
 BOOST_AUTO_TEST_SUITE(rlpx)
@@ -451,5 +453,55 @@ BOOST_AUTO_TEST_CASE(ecies_interop_test_primitives)
 	BOOST_REQUIRE(plainTest3 == expectedPlain3);
 }
 
+BOOST_AUTO_TEST_CASE(readerWriter)
+{
+	ECDHE localEph;
+	h256 localNonce = Nonce::get();
+	ECDHE remoteEph;
+	h256 remoteNonce = Nonce::get();
+	bytes ackCipher{0};
+	bytes authCipher{1};
+	RLPXFrameCoder coder(true, remoteEph.pubkey(), remoteNonce, localEph, localNonce, &ackCipher, &authCipher);
+	
+	/// Test writing a 64byte packet and drain with minimum frame size that
+	/// forces packet to be pieced into 4 frames.
+	/// (Minimum frame size has room for 16 bytes of payload)
+	
+	// 64-byte payload minus 3 bytes for packet-type and RLP overhead.
+	auto dequeLen = 16;
+	bytes stuff = sha3("A").asBytes();
+	bytes payload;
+	payload += stuff;
+	payload += stuff;
+	payload.resize(payload.size() - 3 /* packet-type, rlp-overhead */);
+	BOOST_REQUIRE_EQUAL(61, payload.size());
+	
+	auto drains = (payload.size() + 3) / dequeLen;
+	BOOST_REQUIRE_EQUAL(4, drains);
+	
+	RLPXFrameWriter w(0);
+	w.enque(0, (RLPStream() << payload));
+	vector<bytes> out;
+	for (unsigned i = 1; i < drains; i++)
+	{
+		auto n = w.drain(coder, RLPXFrameWriter::MinFrameDequeLength, out);
+		BOOST_REQUIRE_EQUAL(0, n);
+		BOOST_REQUIRE_EQUAL(out.size(), i);
+	}
+	BOOST_REQUIRE_EQUAL(1, w.drain(coder, RLPXFrameWriter::MinFrameDequeLength, out));
+	BOOST_REQUIRE_EQUAL(out.size(), drains);
+	BOOST_REQUIRE_EQUAL(0, w.drain(coder, RLPXFrameWriter::MinFrameDequeLength, out));
+	BOOST_REQUIRE_EQUAL(out.size(), drains);
+	
+	// we should now have a bunch of ciphertext in out
+	BOOST_REQUIRE(out.size() == drains);
+	for (auto const& c: out)
+		BOOST_REQUIRE(c.size() == RLPXFrameWriter::MinFrameDequeLength);
+	
+	// read and assemble dequed frames
+	
+	
+}
+
 BOOST_AUTO_TEST_SUITE_END()