Message.cs at 94e500b ⋅ planetarium/libplanet

using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.IO;
using System.Linq;
using System.Runtime.Serialization.Formatters.Binary;
using Libplanet.Blocks;
using Libplanet.Crypto;
using NetMQ;

namespace Libplanet.Net.Messages
{
    internal abstract class Message
    {
        public const int CommonFrames = 4;

        internal enum MessageType : byte
        {
            /// <summary>
            /// Check message to determine peer is alive.
            /// </summary>
            Ping = 0x01,

            /// <summary>
            /// A reply to <see cref="Ping"/>.
            /// </summary>
            Pong = 0x14,

            /// <summary>
            /// Request to query block hashes.
            /// </summary>
            GetBlockHashes = 0x04,

            /// <summary>
            /// Inventory to transfer transactions.
            /// </summary>
            TxIds = 0x06,

            /// <summary>
            /// Request to query blocks.
            /// </summary>
            GetBlocks = 0x07,

            /// <summary>
            /// Request to query transactions.
            /// </summary>
            GetTxs = 0x08,

            /// <summary>
            /// Message containing serialized blocks.
            /// </summary>
            Blocks = 0x0a,

            /// <summary>
            /// Message containing serialized transaction.
            /// </summary>
            Tx = 0x10,

            /// <summary>
            /// Message containing request for nearby peers.
            /// </summary>
            FindNeighbors = 0x11,

            /// <summary>
            /// Message containing nearby peers.
            /// </summary>
            Neighbors = 0x12,

            /// <summary>
            /// Request to query calculated states.
            /// </summary>
            GetRecentStates = 0x0b,

            /// <summary>
            /// A reply to <see cref="GetRecentStates"/>.
            /// Contains the calculated recent states and state references.
            /// </summary>
            RecentStates = 0x13,

            /// <summary>
            /// Message containing a single <see cref="BlockHeader"/>.
            /// </summary>
            BlockHeaderMessage = 0x0c,

            /// <summary>
            /// Message containing demand block hashes with their index numbers.
            /// </summary>
            BlockHashes = 0x0e,

            /// <summary>
            /// Request current chain status of the peer.
            /// </summary>
            GetChainStatus = 0x20,

            /// <summary>
            /// A reply to <see cref="GetChainStatus"/>.
            /// Contains the chain status of the peer at the moment.
            /// </summary>
            ChainStatus = 0x24,

            /// <summary>
            /// Request a block's delta states.
            /// </summary>
            GetBlockStates = 0x22,

            /// <summary>
            /// A reply to <see cref="GetBlockStates"/>.
            /// Contains the delta states of the requested block.
            /// </summary>
            BlockStates = 0x23,

            /// <summary>
            /// A reply to any messages with different <see cref="AppProtocolVersion"/>.
            /// Contains the expected and actual <see cref="AppProtocolVersion"/>
            /// value of the message.
            /// </summary>
            DifferentVersion = 0x30,
        }

        private enum MessageFrame
        {
            /// <summary>
            /// Frame containing <see cref="AppProtocolVersion"/>.
            /// </summary>
            Version = 0,

            /// <summary>
            /// Frame containing <see cref="MessageType"/>.
            /// </summary>
            Type = 1,

            /// <summary>
            /// Frame containing the sender <see cref="Peer"/> of the <see cref="Message"/>.
            /// </summary>
            Peer = 2,

            /// <summary>
            /// Frame containing signature of the <see cref="Message"/>.
            /// </summary>
            Sign = 3,
        }

        public byte[] Identity { get; set; }

        public AppProtocolVersion Version { get; set; }

        public Peer Remote { get; set; }

        protected abstract MessageType Type { get; }

        protected abstract IEnumerable<NetMQFrame> DataFrames { get; }

        public static Message Parse(
            NetMQMessage raw,
            bool reply,
            AppProtocolVersion localVersion,
            IImmutableSet<PublicKey> trustedAppProtocolVersionSigners,
            DifferentAppProtocolVersionEncountered differentAppProtocolVersionEncountered)
        {
            if (raw.FrameCount == 0)
            {
                throw new ArgumentException("Can't parse empty NetMQMessage.");
            }

            // (reply == true)  [version, type, peer, sign, frames...]
            // (reply == false) [identity, version, type, peer, sign, frames...]
            NetMQFrame[] remains = reply ? raw.ToArray() : raw.Skip(1).ToArray();

            var versionToken = remains[(int)MessageFrame.Version].ConvertToString();

            AppProtocolVersion remoteVersion = AppProtocolVersion.FromToken(versionToken);
            Peer remotePeer = null;
            try
            {
                remotePeer = DeserializePeer(remains[(int)MessageFrame.Peer].ToByteArray());
            }
            catch (Exception)
            {
                // If failed to find out remotePeer, leave it null.
            }

            if (!IsAppProtocolVersionValid(
                remotePeer,
                localVersion,
                remoteVersion,
                trustedAppProtocolVersionSigners,
                differentAppProtocolVersionEncountered))
            {
                throw new DifferentAppProtocolVersionException(
                    "Received message's version is not valid.",
                    reply ? null : raw[0].Buffer.ToArray(),
                    localVersion,
                    remoteVersion);
            }

            var rawType = (MessageType)remains[(int)MessageFrame.Type].ConvertToInt32();
            var peer = remains[(int)MessageFrame.Peer].ToByteArray();
            byte[] signature = remains[(int)MessageFrame.Sign].ToByteArray();

            NetMQFrame[] body = remains.Skip(CommonFrames).ToArray();

            // FIXME: The below code is too repetitive and prone to miss to add, which means,
            // when you add a new message type, you adds an enum member to MessageType and
            // a corresponding subclass of Message, but misses to add that correspondence here,
            // you may take a long time to be aware you've missed here, because the code is still
            // built well and it looks like just Swarm<T> silently ignore new messages.
            // At least this correspondence map should not be here.
            var types = new Dictionary<MessageType, Type>
            {
                { MessageType.Ping, typeof(Ping) },
                { MessageType.Pong, typeof(Pong) },
                { MessageType.GetBlockHashes, typeof(GetBlockHashes) },
                { MessageType.BlockHashes, typeof(BlockHashes) },
                { MessageType.TxIds, typeof(TxIds) },
                { MessageType.GetBlocks, typeof(GetBlocks) },
                { MessageType.GetTxs, typeof(GetTxs) },
                { MessageType.Blocks, typeof(Blocks) },
                { MessageType.Tx, typeof(Tx) },
                { MessageType.FindNeighbors, typeof(FindNeighbors) },
                { MessageType.Neighbors, typeof(Neighbors) },
                { MessageType.GetRecentStates, typeof(GetRecentStates) },
                { MessageType.RecentStates, typeof(RecentStates) },
                { MessageType.BlockHeaderMessage, typeof(BlockHeaderMessage) },
                { MessageType.GetChainStatus, typeof(GetChainStatus) },
                { MessageType.ChainStatus, typeof(ChainStatus) },
                { MessageType.GetBlockStates, typeof(GetBlockStates) },
                { MessageType.BlockStates, typeof(BlockStates) },
                { MessageType.DifferentVersion, typeof(DifferentVersion) },
            };

            if (!types.TryGetValue(rawType, out Type type))
            {
                throw new ArgumentException(
                    $"Can't determine NetMQMessage. [type: {rawType}]",
                    nameof(raw)
                );
            }

            var message = (Message)Activator.CreateInstance(
                type, new[] { body });
            message.Version = remoteVersion;
            message.Remote = remotePeer;

            if (!message.Remote.PublicKey.Verify(body.ToByteArray(), signature))
            {
                throw new InvalidMessageException("The message signature is invalid", message);
            }

            if (!reply)
            {
                message.Identity = raw[0].Buffer.ToArray();
            }

            return message;
        }

        public NetMQMessage ToNetMQMessage(PrivateKey key, Peer peer, AppProtocolVersion version)
        {
            if (peer is null)
            {
                throw new ArgumentNullException(nameof(peer));
            }

            var message = new NetMQMessage();

            // Write body (by concrete class)
            foreach (NetMQFrame frame in DataFrames)
            {
                message.Append(frame);
            }

            // Write headers. (inverse order)
            message.Push(key.Sign(message.ToByteArray()));
            message.Push(SerializePeer(peer));
            message.Push((byte)Type);
            message.Push(version.Token);

            if (Identity is byte[] to)
            {
                message.Push(to);
            }

            return message;
        }

        protected static Peer DeserializePeer(byte[] bytes)
        {
            var formatter = new BinaryFormatter();
            using MemoryStream stream = new MemoryStream(bytes);
            return (Peer)formatter.Deserialize(stream);
        }

        protected byte[] SerializePeer(Peer peer)
        {
            var formatter = new BinaryFormatter();
            using MemoryStream stream = new MemoryStream();
            formatter.Serialize(stream, peer);
            return stream.ToArray();
        }

        private static bool IsAppProtocolVersionValid(
            Peer remotePeer,
            AppProtocolVersion localVersion,
            AppProtocolVersion remoteVersion,
            IImmutableSet<PublicKey> trustedAppProtocolVersionSigners,
            DifferentAppProtocolVersionEncountered differentAppProtocolVersionEncountered)
        {
            if (remoteVersion.Equals(localVersion))
            {
                return true;
            }

            if (!(trustedAppProtocolVersionSigners is null) &&
                !trustedAppProtocolVersionSigners.Any(remoteVersion.Verify))
            {
                return false;
            }

            if (differentAppProtocolVersionEncountered is null)
            {
                return false;
            }

            return differentAppProtocolVersionEncountered(remotePeer, remoteVersion, localVersion);
        }
    }
}

Read our documentation on viewing source code .

Navigation	Overlay
`t` Navigate files	`h` Toggle hits
`y` Change url to tip of branch	`m` Toggle misses
`b / v` Jump to prev/next hit line	`p` Toggle partial
`z / x` Jump to prev/next missed or partial line	`1..9` Toggle flags
`shift + o` Open current page in GitHub	`a` Toggle all on
`/ or ?` Show keyboard shortcuts dialog	`c` Toggle context lines or commits

1		using System;
2		using System.Collections.Generic;
3		using System.Collections.Immutable;
4		using System.IO;
5		using System.Linq;
6		using System.Runtime.Serialization.Formatters.Binary;
7		using Libplanet.Blocks;
8		using Libplanet.Crypto;
9		using NetMQ;
10
11		namespace Libplanet.Net.Messages
12		{
13		internal abstract class Message
14		{
15		public const int CommonFrames = 4;
16
17		internal enum MessageType : byte
18		{
19		/// <summary>
20		/// Check message to determine peer is alive.
21		/// </summary>
22		Ping = 0x01,
23
24		/// <summary>
25		/// A reply to <see cref="Ping"/>.
26		/// </summary>
27		Pong = 0x14,
28
29		/// <summary>
30		/// Request to query block hashes.
31		/// </summary>
32		GetBlockHashes = 0x04,
33
34		/// <summary>
35		/// Inventory to transfer transactions.
36		/// </summary>
37		TxIds = 0x06,
38
39		/// <summary>
40		/// Request to query blocks.
41		/// </summary>
42		GetBlocks = 0x07,
43
44		/// <summary>
45		/// Request to query transactions.
46		/// </summary>
47		GetTxs = 0x08,
48
49		/// <summary>
50		/// Message containing serialized blocks.
51		/// </summary>
52		Blocks = 0x0a,
53
54		/// <summary>
55		/// Message containing serialized transaction.
56		/// </summary>
57		Tx = 0x10,
58
59		/// <summary>
60		/// Message containing request for nearby peers.
61		/// </summary>
62		FindNeighbors = 0x11,
63
64		/// <summary>
65		/// Message containing nearby peers.
66		/// </summary>
67		Neighbors = 0x12,
68
69		/// <summary>
70		/// Request to query calculated states.
71		/// </summary>
72		GetRecentStates = 0x0b,
73
74		/// <summary>
75		/// A reply to <see cref="GetRecentStates"/>.
76		/// Contains the calculated recent states and state references.
77		/// </summary>
78		RecentStates = 0x13,
79
80		/// <summary>
81		/// Message containing a single <see cref="BlockHeader"/>.
82		/// </summary>
83		BlockHeaderMessage = 0x0c,
84
85		/// <summary>
86		/// Message containing demand block hashes with their index numbers.
87		/// </summary>
88		BlockHashes = 0x0e,
89
90		/// <summary>
91		/// Request current chain status of the peer.
92		/// </summary>
93		GetChainStatus = 0x20,
94
95		/// <summary>
96		/// A reply to <see cref="GetChainStatus"/>.
97		/// Contains the chain status of the peer at the moment.
98		/// </summary>
99		ChainStatus = 0x24,
100
101		/// <summary>
102		/// Request a block's delta states.
103		/// </summary>
104		GetBlockStates = 0x22,
105
106		/// <summary>
107		/// A reply to <see cref="GetBlockStates"/>.
108		/// Contains the delta states of the requested block.
109		/// </summary>
110		BlockStates = 0x23,
111
112		/// <summary>
113		/// A reply to any messages with different <see cref="AppProtocolVersion"/>.
114		/// Contains the expected and actual <see cref="AppProtocolVersion"/>
115		/// value of the message.
116		/// </summary>
117		DifferentVersion = 0x30,
118		}
119
120		private enum MessageFrame
121		{
122		/// <summary>
123		/// Frame containing <see cref="AppProtocolVersion"/>.
124		/// </summary>
125		Version = 0,
126
127		/// <summary>
128		/// Frame containing <see cref="MessageType"/>.
129		/// </summary>
130		Type = 1,
131
132		/// <summary>
133		/// Frame containing the sender <see cref="Peer"/> of the <see cref="Message"/>.
134		/// </summary>
135		Peer = 2,
136
137		/// <summary>
138		/// Frame containing signature of the <see cref="Message"/>.
139		/// </summary>
140		Sign = 3,
141		}
142
143	1	public byte[] Identity { get; set; }
144
145	1	public AppProtocolVersion Version { get; set; }
146
147	1	public Peer Remote { get; set; }
148
149		protected abstract MessageType Type { get; }
150
151		protected abstract IEnumerable<NetMQFrame> DataFrames { get; }
152
153		public static Message Parse(
154		NetMQMessage raw,
155		bool reply,
156		AppProtocolVersion localVersion,
157		IImmutableSet<PublicKey> trustedAppProtocolVersionSigners,
158		DifferentAppProtocolVersionEncountered differentAppProtocolVersionEncountered)
159	1	{
160	1	if (raw.FrameCount == 0)
161	0	{
162	0	throw new ArgumentException("Can't parse empty NetMQMessage.");
163		}
164
165		// (reply == true) [version, type, peer, sign, frames...]
166		// (reply == false) [identity, version, type, peer, sign, frames...]
167	1	NetMQFrame[] remains = reply ? raw.ToArray() : raw.Skip(1).ToArray();
168
169	1	var versionToken = remains[(int)MessageFrame.Version].ConvertToString();
170
171	1	AppProtocolVersion remoteVersion = AppProtocolVersion.FromToken(versionToken);
172	1	Peer remotePeer = null;
173		try
174	1	{
175	1	remotePeer = DeserializePeer(remains[(int)MessageFrame.Peer].ToByteArray());
176	1	}
177	1	catch (Exception)
178	1	{
179		// If failed to find out remotePeer, leave it null.
180	1	}
181
182	1	if (!IsAppProtocolVersionValid(
183	1	remotePeer,
184	1	localVersion,
185	1	remoteVersion,
186	1	trustedAppProtocolVersionSigners,
187	1	differentAppProtocolVersionEncountered))
188	1	{
189	1	throw new DifferentAppProtocolVersionException(
190	1	"Received message's version is not valid.",
191	1	reply ? null : raw[0].Buffer.ToArray(),
192	1	localVersion,
193	1	remoteVersion);
194		}
195
196	1	var rawType = (MessageType)remains[(int)MessageFrame.Type].ConvertToInt32();
197	1	var peer = remains[(int)MessageFrame.Peer].ToByteArray();
198	1	byte[] signature = remains[(int)MessageFrame.Sign].ToByteArray();
199
200	1	NetMQFrame[] body = remains.Skip(CommonFrames).ToArray();
201
202		// FIXME: The below code is too repetitive and prone to miss to add, which means,
203		// when you add a new message type, you adds an enum member to MessageType and
204		// a corresponding subclass of Message, but misses to add that correspondence here,
205		// you may take a long time to be aware you've missed here, because the code is still
206		// built well and it looks like just Swarm<T> silently ignore new messages.
207		// At least this correspondence map should not be here.
208	1	var types = new Dictionary<MessageType, Type>
209	1	{
210	1	{ MessageType.Ping, typeof(Ping) },
211	1	{ MessageType.Pong, typeof(Pong) },
212	1	{ MessageType.GetBlockHashes, typeof(GetBlockHashes) },
213	1	{ MessageType.BlockHashes, typeof(BlockHashes) },
214	1	{ MessageType.TxIds, typeof(TxIds) },
215	1	{ MessageType.GetBlocks, typeof(GetBlocks) },
216	1	{ MessageType.GetTxs, typeof(GetTxs) },
217	1	{ MessageType.Blocks, typeof(Blocks) },
218	1	{ MessageType.Tx, typeof(Tx) },
219	1	{ MessageType.FindNeighbors, typeof(FindNeighbors) },
220	1	{ MessageType.Neighbors, typeof(Neighbors) },
221	1	{ MessageType.GetRecentStates, typeof(GetRecentStates) },
222	1	{ MessageType.RecentStates, typeof(RecentStates) },
223	1	{ MessageType.BlockHeaderMessage, typeof(BlockHeaderMessage) },
224	1	{ MessageType.GetChainStatus, typeof(GetChainStatus) },
225	1	{ MessageType.ChainStatus, typeof(ChainStatus) },
226	1	{ MessageType.GetBlockStates, typeof(GetBlockStates) },
227	1	{ MessageType.BlockStates, typeof(BlockStates) },
228	1	{ MessageType.DifferentVersion, typeof(DifferentVersion) },
229	1	};
230
231	1	if (!types.TryGetValue(rawType, out Type type))
232	0	{
233	0	throw new ArgumentException(
234	0	$"Can't determine NetMQMessage. [type: {rawType}]",
235	0	nameof(raw)
236	0	);
237		}
238
239	1	var message = (Message)Activator.CreateInstance(
240	1	type, new[] { body });
241	1	message.Version = remoteVersion;
242	1	message.Remote = remotePeer;
243
244	1	if (!message.Remote.PublicKey.Verify(body.ToByteArray(), signature))
245	0	{
246	0	throw new InvalidMessageException("The message signature is invalid", message);
247		}
248
249	1	if (!reply)
250	1	{
251	1	message.Identity = raw[0].Buffer.ToArray();
252	1	}
253
254	1	return message;
255	1	}
256
257		public NetMQMessage ToNetMQMessage(PrivateKey key, Peer peer, AppProtocolVersion version)
258	1	{
259	1	if (peer is null)
260	0	{
261	0	throw new ArgumentNullException(nameof(peer));
262		}
263
264	1	var message = new NetMQMessage();
265
266		// Write body (by concrete class)
267	1	foreach (NetMQFrame frame in DataFrames)
268	1	{
269	1	message.Append(frame);
270	1	}
271
272		// Write headers. (inverse order)
273	1	message.Push(key.Sign(message.ToByteArray()));
274	1	message.Push(SerializePeer(peer));
275	1	message.Push((byte)Type);
276	1	message.Push(version.Token);
277
278	1	if (Identity is byte[] to)
279	1	{
280	1	message.Push(to);
281	1	}
282
283	1	return message;
284	1	}
285
286		protected static Peer DeserializePeer(byte[] bytes)
287	1	{
288	1	var formatter = new BinaryFormatter();
289	1	using MemoryStream stream = new MemoryStream(bytes);
290	1	return (Peer)formatter.Deserialize(stream);
291	1	}
292
293		protected byte[] SerializePeer(Peer peer)
294	1	{
295	1	var formatter = new BinaryFormatter();
296	1	using MemoryStream stream = new MemoryStream();
297	1	formatter.Serialize(stream, peer);
298	1	return stream.ToArray();
299	1	}
300
301		private static bool IsAppProtocolVersionValid(
302		Peer remotePeer,
303		AppProtocolVersion localVersion,
304		AppProtocolVersion remoteVersion,
305		IImmutableSet<PublicKey> trustedAppProtocolVersionSigners,
306		DifferentAppProtocolVersionEncountered differentAppProtocolVersionEncountered)
307	1	{
308	1	if (remoteVersion.Equals(localVersion))
309	1	{
310	1	return true;
311		}
312
313	1	if (!(trustedAppProtocolVersionSigners is null) &&
314	1	!trustedAppProtocolVersionSigners.Any(remoteVersion.Verify))
315	1	{
316	1	return false;
317		}
318
319	1	if (differentAppProtocolVersionEncountered is null)
320	1	{
321	1	return false;
322		}
323
324	1	return differentAppProtocolVersionEncountered(remotePeer, remoteVersion, localVersion);
325	1	}
326		}
327		}