net/message: optional max message bytes limit added

bin/darkfid/src/proto/protocol_proposal.rs

@@ -46,7 +46,7 @@ use crate::task::handle_unknown_proposal;
 #[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
 pub struct ProposalMessage(pub Proposal);
 
-impl_p2p_message!(ProposalMessage, "proposal");
+impl_p2p_message!(ProposalMessage, "proposal", 0);
 
 /// Atomic pointer to the `ProtocolProposal` handler.
 pub type ProtocolProposalHandlerPtr = Arc<ProtocolProposalHandler>;

bin/darkfid/src/proto/protocol_sync.rs

@@ -49,7 +49,7 @@ pub struct TipRequest {
     pub tip: HeaderHash,
 }
 
-impl_p2p_message!(TipRequest, "tiprequest");
+impl_p2p_message!(TipRequest, "tiprequest", 0);
 
 /// Structure representing the response to `TipRequest`,
 /// containing a boolean flag to indicate if we are synced,
@@ -64,7 +64,7 @@ pub struct TipResponse {
     pub hash: Option<HeaderHash>,
 }
 
-impl_p2p_message!(TipResponse, "tipresponse");
+impl_p2p_message!(TipResponse, "tipresponse", 0);
 
 /// Structure represening a request to ask a node for up to `BATCH` headers before
 /// the provided header height.
@@ -74,7 +74,7 @@ pub struct HeaderSyncRequest {
     pub height: u32,
 }
 
-impl_p2p_message!(HeaderSyncRequest, "headersyncrequest");
+impl_p2p_message!(HeaderSyncRequest, "headersyncrequest", 0);
 
 /// Structure representing the response to `HeaderSyncRequest`,
 /// containing up to `BATCH` headers before the requested block height.
@@ -84,7 +84,7 @@ pub struct HeaderSyncResponse {
     pub headers: Vec<Header>,
 }
 
-impl_p2p_message!(HeaderSyncResponse, "headersyncresponse");
+impl_p2p_message!(HeaderSyncResponse, "headersyncresponse", 0);
 
 /// Structure represening a request to ask a node for up to`BATCH` blocks
 /// of provided headers.
@@ -94,7 +94,7 @@ pub struct SyncRequest {
     pub headers: Vec<HeaderHash>,
 }
 
-impl_p2p_message!(SyncRequest, "syncrequest");
+impl_p2p_message!(SyncRequest, "syncrequest", 0);
 
 /// Structure representing the response to `SyncRequest`,
 /// containing up to `BATCH` blocks after the requested block height.
@@ -104,7 +104,7 @@ pub struct SyncResponse {
     pub blocks: Vec<BlockInfo>,
 }
 
-impl_p2p_message!(SyncResponse, "syncresponse");
+impl_p2p_message!(SyncResponse, "syncresponse", 0);
 
 /// Structure represening a request to ask a node a fork sequence.
 /// If we include a specific fork tip, they have to return its sequence,
@@ -119,7 +119,7 @@ pub struct ForkSyncRequest {
     pub fork_tip: Option<HeaderHash>,
 }
 
-impl_p2p_message!(ForkSyncRequest, "forksyncrequest");
+impl_p2p_message!(ForkSyncRequest, "forksyncrequest", 0);
 
 /// Structure representing the response to `ForkSyncRequest`,
 /// containing the requested fork sequence.
@@ -129,7 +129,7 @@ pub struct ForkSyncResponse {
     pub proposals: Vec<Proposal>,
 }
 
-impl_p2p_message!(ForkSyncResponse, "forksyncresponse");
+impl_p2p_message!(ForkSyncResponse, "forksyncresponse", 0);
 
 /// Structure represening a request to ask a node a fork header for the
 /// requested height. The fork is identified by the provided header hash.
@@ -141,7 +141,7 @@ pub struct ForkHeaderHashRequest {
     pub fork_header: HeaderHash,
 }
 
-impl_p2p_message!(ForkHeaderHashRequest, "forkheaderhashrequest");
+impl_p2p_message!(ForkHeaderHashRequest, "forkheaderhashrequest", 0);
 
 /// Structure representing the response to `ForkHeaderHashRequest`,
 /// containing the requested fork header hash, if it was found.
@@ -151,7 +151,7 @@ pub struct ForkHeaderHashResponse {
     pub fork_header: Option<HeaderHash>,
 }
 
-impl_p2p_message!(ForkHeaderHashResponse, "forkheaderhashresponse");
+impl_p2p_message!(ForkHeaderHashResponse, "forkheaderhashresponse", 0);
 
 /// Structure represening a request to ask a node for up to `BATCH`
 /// fork headers for provided header hashes.  The fork is identified
@@ -164,7 +164,7 @@ pub struct ForkHeadersRequest {
     pub fork_header: HeaderHash,
 }
 
-impl_p2p_message!(ForkHeadersRequest, "forkheadersrequest");
+impl_p2p_message!(ForkHeadersRequest, "forkheadersrequest", 0);
 
 /// Structure representing the response to `ForkHeadersRequest`,
 /// containing up to `BATCH` fork headers.
@@ -174,7 +174,7 @@ pub struct ForkHeadersResponse {
     pub headers: Vec<Header>,
 }
 
-impl_p2p_message!(ForkHeadersResponse, "forkheadersresponse");
+impl_p2p_message!(ForkHeadersResponse, "forkheadersresponse", 0);
 
 /// Structure represening a request to ask a node for up to `BATCH`
 /// fork proposals for provided header hashes.  The fork is identified
@@ -187,7 +187,7 @@ pub struct ForkProposalsRequest {
     pub fork_header: HeaderHash,
 }
 
-impl_p2p_message!(ForkProposalsRequest, "forkproposalsrequest");
+impl_p2p_message!(ForkProposalsRequest, "forkproposalsrequest", 0);
 
 /// Structure representing the response to `ForkProposalsRequest`,
 /// containing up to `BATCH` fork headers.
@@ -197,7 +197,7 @@ pub struct ForkProposalsResponse {
     pub proposals: Vec<Proposal>,
 }
 
-impl_p2p_message!(ForkProposalsResponse, "forkproposalsresponse");
+impl_p2p_message!(ForkProposalsResponse, "forkproposalsresponse", 0);
 
 /// Atomic pointer to the `ProtocolSync` handler.
 pub type ProtocolSyncHandlerPtr = Arc<ProtocolSyncHandler>;

bin/dhtd/dhtd/src/proto.rs

@@ -53,7 +53,7 @@ pub struct ChunkRequest {
     pub hash: blake3::Hash,
 }
 
-impl_p2p_message!(ChunkRequest, "dhtchunkrequest");
+impl_p2p_message!(ChunkRequest, "dhtchunkrequest", 0);
 
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct ChunkReply {
@@ -61,14 +61,14 @@ pub struct ChunkReply {
     pub data: Vec<u8>,
 }
 
-impl_p2p_message!(ChunkReply, "dhtchunkreply");
+impl_p2p_message!(ChunkReply, "dhtchunkreply", 0);
 
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FileRequest {
     pub hash: blake3::Hash,
 }
 
-impl_p2p_message!(FileRequest, "dhtfilerequest");
+impl_p2p_message!(FileRequest, "dhtfilerequest", 0);
 
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FileReply {
@@ -76,7 +76,7 @@ pub struct FileReply {
     pub chunks: Vec<blake3::Hash>,
 }
 
-impl_p2p_message!(FileReply, "dhtfilereply");
+impl_p2p_message!(FileReply, "dhtfilereply", 0);
 
 impl ProtocolDht {
     #[allow(dead_code)]

bin/fud/fud/src/proto.rs

@@ -41,14 +41,14 @@ pub struct FudFilePut {
     pub file_hash: blake3::Hash,
     pub chunk_hashes: Vec<blake3::Hash>,
 }
-impl_p2p_message!(FudFilePut, "FudFilePut");
+impl_p2p_message!(FudFilePut, "FudFilePut", 0);
 
 /// Message representing a new chunk on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudChunkPut {
     pub chunk_hash: blake3::Hash,
 }
-impl_p2p_message!(FudChunkPut, "FudChunkPut");
+impl_p2p_message!(FudChunkPut, "FudChunkPut", 0);
 
 /// Message representing a new route for a file on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
@@ -57,7 +57,7 @@ pub struct FudFileRoute {
     pub chunk_hashes: Vec<blake3::Hash>,
     pub peer: Url,
 }
-impl_p2p_message!(FudFileRoute, "FudFileRoute");
+impl_p2p_message!(FudFileRoute, "FudFileRoute", 0);
 
 /// Message representing a new route for a chunk on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
@@ -65,28 +65,28 @@ pub struct FudChunkRoute {
     pub chunk_hash: blake3::Hash,
     pub peer: Url,
 }
-impl_p2p_message!(FudChunkRoute, "FudChunkRoute");
+impl_p2p_message!(FudChunkRoute, "FudChunkRoute", 0);
 
 /// Message representing a file request from the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudFileRequest {
     pub file_hash: blake3::Hash,
 }
-impl_p2p_message!(FudFileRequest, "FudFileRequest");
+impl_p2p_message!(FudFileRequest, "FudFileRequest", 0);
 
 /// Message representing a file reply from the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudFileReply {
     pub chunk_hashes: Vec<blake3::Hash>,
 }
-impl_p2p_message!(FudFileReply, "FudFileReply");
+impl_p2p_message!(FudFileReply, "FudFileReply", 0);
 
 /// Message representing a chunk request from the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudChunkRequest {
     pub chunk_hash: blake3::Hash,
 }
-impl_p2p_message!(FudChunkRequest, "FudChunkRequest");
+impl_p2p_message!(FudChunkRequest, "FudChunkRequest", 0);
 
 /// Message representing a chunk reply from the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
@@ -94,17 +94,17 @@ pub struct FudChunkReply {
     // TODO: This sould be a chunk-sized array, but then we need padding?
     pub chunk: Vec<u8>,
 }
-impl_p2p_message!(FudChunkReply, "FudChunkReply");
+impl_p2p_message!(FudChunkReply, "FudChunkReply", 0);
 
 /// Message representing a chunk reply when a file is not found
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudFileNotFound;
-impl_p2p_message!(FudFileNotFound, "FudFileNotFound");
+impl_p2p_message!(FudFileNotFound, "FudFileNotFound", 0);
 
 /// Message representing a chunk reply when a chunk is not found
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudChunkNotFound;
-impl_p2p_message!(FudChunkNotFound, "FudChunkNotFound");
+impl_p2p_message!(FudChunkNotFound, "FudChunkNotFound", 0);
 
 /// P2P protocol implementation for fud.
 pub struct ProtocolFud {

example/dchat/dchatd/src/dchatmsg.rs

@@ -30,5 +30,5 @@ pub struct DchatMsg {
     pub msg: String,
 }
 
-impl_p2p_message!(DchatMsg, "DchatMsg");
+impl_p2p_message!(DchatMsg, "DchatMsg", 0);
 // ANCHOR_END: msg

script/research/dam/dam-localnet/damd0.toml

@@ -26,7 +26,7 @@ hostlist = "damd0/p2p_hostlist.tsv"
 inbound = ["tcp+tls://0.0.0.0:44781"]
 
 # Peer nodes to manually connect to
-peers = ["tcp+tls://0.0.0.0:44881"]
+peers = []
 
 # Whitelisted network transports for outbound connections
 allowed_transports = ["tcp+tls"]

script/research/dam/damd/src/proto/protocol_bar.rs

@@ -44,7 +44,7 @@ pub struct Bar {
     pub message: String,
 }
 
-impl_p2p_message!(Bar, "bar");
+impl_p2p_message!(Bar, "bar", 0);
 
 /// Atomic pointer to the `ProtocolBar` handler.
 pub type ProtocolBarHandlerPtr = Arc<ProtocolBarHandler>;

script/research/dam/damd/src/proto/protocol_foo.rs

@@ -44,7 +44,7 @@ pub struct FooRequest {
     pub message: String,
 }
 
-impl_p2p_message!(FooRequest, "foorequest");
+impl_p2p_message!(FooRequest, "foorequest", 0);
 
 /// Structure representing the response to `FooRequest`.
 #[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
@@ -53,7 +53,7 @@ pub struct FooResponse {
     pub code: u8,
 }
 
-impl_p2p_message!(FooResponse, "fooresponse");
+impl_p2p_message!(FooResponse, "fooresponse", 0);
 
 /// Atomic pointer to the `ProtocolFoo` handler.
 pub type ProtocolFooHandlerPtr = Arc<ProtocolFooHandler>;

script/research/generic-node/src/main.rs

@@ -71,25 +71,25 @@ struct Args {
 struct GenericStringMessage {
     msg: String,
 }
-impl_p2p_message!(GenericStringMessage, "generic_string_message");
+impl_p2p_message!(GenericStringMessage, "generic_string_message", 0);
 
 #[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
 struct GenericNumberMessage {
     num: u64,
 }
-impl_p2p_message!(GenericNumberMessage, "generic_number_message");
+impl_p2p_message!(GenericNumberMessage, "generic_number_message", 0);
 
 #[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
 struct GenericRequestMessage {
     msg: String,
 }
-impl_p2p_message!(GenericRequestMessage, "generic_request_message");
+impl_p2p_message!(GenericRequestMessage, "generic_request_message", 0);
 
 #[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
 struct GenericResponseMessage {
     msg: String,
 }
-impl_p2p_message!(GenericResponseMessage, "generic_response_message");
+impl_p2p_message!(GenericResponseMessage, "generic_response_message", 0);
 
 /// Generic daemon structure
 struct Genericd {

src/error.rs

@@ -155,6 +155,9 @@ pub enum Error {
     #[error("Missing P2P message dispatcher")]
     MissingDispatcher,
 
+    #[error("P2P message is invalid")]
+    MessageInvalid,
+
     #[cfg(feature = "arti-client")]
     #[error(transparent)]
     ArtiError(#[from] arti_client::Error),

src/event_graph/proto.rs

@@ -116,27 +116,27 @@ pub struct ProtocolEventGraph {
 /// A P2P message representing publishing an event on the network
 #[derive(Clone, SerialEncodable, SerialDecodable)]
 pub struct EventPut(pub Event);
-impl_p2p_message!(EventPut, "EventGraph::EventPut");
+impl_p2p_message!(EventPut, "EventGraph::EventPut", 0);
 
 /// A P2P message representing an event request
 #[derive(Clone, SerialEncodable, SerialDecodable)]
 pub struct EventReq(pub Vec<blake3::Hash>);
-impl_p2p_message!(EventReq, "EventGraph::EventReq");
+impl_p2p_message!(EventReq, "EventGraph::EventReq", 0);
 
 /// A P2P message representing an event reply
 #[derive(Clone, SerialEncodable, SerialDecodable)]
 pub struct EventRep(pub Vec<Event>);
-impl_p2p_message!(EventRep, "EventGraph::EventRep");
+impl_p2p_message!(EventRep, "EventGraph::EventRep", 0);
 
 /// A P2P message representing a request for a peer's DAG tips
 #[derive(Clone, SerialEncodable, SerialDecodable)]
 pub struct TipReq {}
-impl_p2p_message!(TipReq, "EventGraph::TipReq");
+impl_p2p_message!(TipReq, "EventGraph::TipReq", 0);
 
 /// A P2P message representing a reply for the peer's DAG tips
 #[derive(Clone, SerialEncodable, SerialDecodable)]
 pub struct TipRep(pub BTreeMap<u64, HashSet<blake3::Hash>>);
-impl_p2p_message!(TipRep, "EventGraph::TipRep");
+impl_p2p_message!(TipRep, "EventGraph::TipRep", 0);
 
 #[async_trait]
 impl ProtocolBase for ProtocolEventGraph {

src/net/channel.rs

@@ -382,9 +382,9 @@ impl Channel {
             // Send result to our publishers
             match self.message_subsystem.notify(&command, reader).await {
                 Ok(()) => {}
-                Err(Error::MissingDispatcher) => {
-                    // If we're getting messages without dispatchers, it's spam.
-                    // We therefore ban this channel if:
+                Err(Error::MissingDispatcher) | Err(Error::MessageInvalid) => {
+                    // If we're getting messages without dispatchers or its invalid,
+                    // it's spam. We therefore ban this channel if:
                     //
                     // 1) This channel is NOT part of a refine session.
                     //

src/net/message.rs

@@ -25,6 +25,9 @@ use url::{Host, Url};
 /// Generic message template.
 pub trait Message: 'static + Send + Sync + AsyncDecodable + AsyncEncodable {
     const NAME: &'static str;
+    /// Message bytes vector length limit.
+    /// Set to 0 for no limit.
+    const MAX_BYTES: u64;
 }
 
 /// Generic serialized message template.
@@ -41,9 +44,10 @@ impl SerializedMessage {
 
 #[macro_export]
 macro_rules! impl_p2p_message {
-    ($st:ty, $nm:expr) => {
+    ($st:ty, $nm:expr, $mb:expr) => {
         impl Message for $st {
             const NAME: &'static str = $nm;
+            const MAX_BYTES: u64 = $mb;
         }
     };
 }
@@ -53,14 +57,14 @@ macro_rules! impl_p2p_message {
 pub struct PingMessage {
     pub nonce: u16,
 }
-impl_p2p_message!(PingMessage, "ping");
+impl_p2p_message!(PingMessage, "ping", 0);
 
 /// Inbound keepalive message.
 #[derive(Debug, Copy, Clone, SerialEncodable, SerialDecodable)]
 pub struct PongMessage {
     pub nonce: u16,
 }
-impl_p2p_message!(PongMessage, "pong");
+impl_p2p_message!(PongMessage, "pong", 0);
 
 /// Requests address of outbound connecction.
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
@@ -73,7 +77,7 @@ pub struct GetAddrsMessage {
     /// Preferred addresses transports
     pub transports: Vec<String>,
 }
-impl_p2p_message!(GetAddrsMessage, "getaddr");
+impl_p2p_message!(GetAddrsMessage, "getaddr", 0);
 
 /// Sends address information to inbound connection.
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
@@ -81,7 +85,7 @@ pub struct AddrsMessage {
     pub addrs: Vec<(Url, u64)>,
 }
 
-impl_p2p_message!(AddrsMessage, "addr");
+impl_p2p_message!(AddrsMessage, "addr", 0);
 
 /// Requests version information of outbound connection.
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
@@ -105,7 +109,7 @@ pub struct VersionMessage {
     /// to be enabled for this connection
     pub features: Vec<(String, u32)>,
 }
-impl_p2p_message!(VersionMessage, "version");
+impl_p2p_message!(VersionMessage, "version", 0);
 
 impl VersionMessage {
     pub(in crate::net) fn get_ipv6_addr(&self) -> Option<Ipv6Addr> {
@@ -125,4 +129,4 @@ pub struct VerackMessage {
     /// App version
     pub app_version: semver::Version,
 }
-impl_p2p_message!(VerackMessage, "verack");
+impl_p2p_message!(VerackMessage, "verack", 0);

src/net/message_publisher.rs

@@ -178,7 +178,10 @@ impl<M: Message> MessageSubscription<M> {
 /// Generic interface for the message dispatcher.
 #[async_trait]
 trait MessageDispatcherInterface: Send + Sync {
-    async fn trigger(&self, stream: &mut smol::io::ReadHalf<Box<dyn PtStream + 'static>>);
+    async fn trigger(
+        &self,
+        stream: &mut smol::io::ReadHalf<Box<dyn PtStream + 'static>>,
+    ) -> Result<()>;
 
     async fn trigger_error(&self, err: Error);
 
@@ -194,37 +197,50 @@ impl<M: Message> MessageDispatcherInterface for MessageDispatcher<M> {
     ///
     /// We extract the message length from the stream and use `take()`
     /// to allocate an appropiately sized buffer as a basic DDOS protection.
-    async fn trigger(&self, stream: &mut smol::io::ReadHalf<Box<dyn PtStream + 'static>>) {
-        match VarInt::decode_async(stream).await {
-            Ok(int) => {
-                // TODO: check the message length does not exceed some bound.
-                let len = int.0;
-                let mut take = stream.take(len);
-
-                // Deserialize stream into type, send down the pipes.
-                match M::decode_async(&mut take).await {
-                    Ok(payload) => {
-                        let message = Ok(Arc::new(payload));
-                        self._trigger_all(message).await
-                    }
-
-                    Err(err) => {
-                        error!(
-                            target: "net::message_publisher::trigger()",
-                            "Unable to decode data. Dropping...: {}",
-                            err,
-                        );
-                    }
-                }
-            }
+    async fn trigger(
+        &self,
+        stream: &mut smol::io::ReadHalf<Box<dyn PtStream + 'static>>,
+    ) -> Result<()> {
+        // Parse message length
+        let length = match VarInt::decode_async(stream).await {
+            Ok(int) => int.0,
             Err(err) => {
                 error!(
                     target: "net::message_publisher::trigger()",
                     "Unable to decode VarInt. Dropping...: {}",
                     err,
                 );
+                return Err(Error::MessageInvalid)
             }
+        };
+
+        // Check the message length does not exceed set limit
+        if M::MAX_BYTES > 0 && length > M::MAX_BYTES {
+            error!(
+                target: "net::message_publisher::trigger()",
+                "Message length ({}) exceeds configured limit ({}). Dropping...",
+                length, M::MAX_BYTES,
+            );
+            return Err(Error::MessageInvalid)
         }
+
+        // Deserialize stream into type
+        let mut take = stream.take(length);
+        let message = match M::decode_async(&mut take).await {
+            Ok(payload) => Ok(Arc::new(payload)),
+            Err(err) => {
+                error!(
+                    target: "net::message_publisher::trigger()",
+                    "Unable to decode data. Dropping...: {}",
+                    err,
+                );
+                return Err(Error::MessageInvalid)
+            }
+        };
+
+        // Send down the pipes
+        self._trigger_all(message).await;
+        Ok(())
     }
 
     /// Internal function that sends an error message to all subscriber channels.
@@ -293,8 +309,7 @@ impl MessageSubsystem {
             return Err(Error::MissingDispatcher)
         };
 
-        dispatcher.trigger(reader).await;
-        Ok(())
+        dispatcher.trigger(reader).await
     }
 
     /// Concurrently transmits an error message across dispatchers.

src/tx/mod.rs

@@ -233,7 +233,7 @@ impl std::fmt::Debug for Transaction {
 use crate::net::Message;
 
 #[cfg(feature = "net")]
-crate::impl_p2p_message!(Transaction, "tx");
+crate::impl_p2p_message!(Transaction, "tx", 0);
 
 /// Calls tree bounds definitions
 // TODO: increase min to 2 when fees are implement