Added bad node identify test

Author: 0xMimir

p2p/src/network/identify/p2p_network_identify_protocol.rs

@@ -77,6 +77,12 @@ impl TryFrom<super::pb::Identify> for P2pNetworkIdentify {
     }
 }
 
+impl P2pNetworkIdentify {
+    pub fn to_proto_message(&self) -> super::pb::Identify {
+        super::pb::Identify::from(self)
+    }
+}
+
 impl<'a> From<&'a P2pNetworkIdentify> for super::pb::Identify {
     fn from(value: &'a P2pNetworkIdentify) -> Self {
         Self {

p2p/src/network/kad/p2p_network_kad_internals.rs

@@ -10,7 +10,8 @@ use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 
 use crate::{
-    ConnectionType, P2pNetworkKademliaMultiaddrError, P2pNetworkKademliaPeerIdError, PeerId,
+    socket_addr_try_from_multiaddr, ConnectionType, P2pNetworkKademliaMultiaddrError,
+    P2pNetworkKademliaPeerIdError, PeerId,
 };
 
 use super::CID;
@@ -245,10 +246,16 @@ impl<const K: usize> P2pNetworkKadRoutingTable<K> {
     /// K-bucket is full and cannot be split), then `Err(_)` value is returned.
     ///
     /// TODO: should it be just another variant in `Ok(_)`?
+    ///
+    /// Filters `entry.addrs` for supported addresses if non of addresses are supported returns `Err(_)`
     pub fn insert(
         &mut self,
-        entry: P2pNetworkKadEntry,
+        mut entry: P2pNetworkKadEntry,
     ) -> Result<bool, P2pNetworkKadRoutingTableInsertError> {
+        entry.filter_addrs();
+        if entry.addrs.is_empty() {
+            return Err(P2pNetworkKadRoutingTableInsertError);
+        }
         // distance to this node
         let dist = &self.this_key - &entry.key;
 
@@ -371,6 +378,18 @@ impl P2pNetworkKadEntry {
     pub fn dist(&self, other: &P2pNetworkKadEntry) -> P2pNetworkKadDist {
         &self.key - &other.key
     }
+
+    pub fn filter_addrs(&mut self) {
+        let addrs = std::mem::take(&mut self.addrs);
+        self.addrs = addrs
+            .into_iter()
+            .filter_map(|multiaddr| {
+                socket_addr_try_from_multiaddr(&multiaddr)
+                    .is_ok()
+                    .then_some(multiaddr)
+            })
+            .collect()
+    }
 }
 
 #[derive(Clone, Debug, Serialize, PartialEq, Deserialize, thiserror::Error)]
@@ -573,6 +592,7 @@ mod tests {
     use std::collections::BTreeSet;
 
     use crypto_bigint::{Random, U256};
+    use multiaddr::multiaddr;
 
     use crate::{identity::SecretKey, PeerId, CID};
 
@@ -601,7 +621,7 @@ mod tests {
         P2pNetworkKadEntry {
             key,
             peer_id,
-            addrs: vec![],
+            addrs: vec![multiaddr!(Ip4([0; 4]), Tcp(1000_u16))],
             connection: super::ConnectionType::Connected,
         }
     }
@@ -611,7 +631,7 @@ mod tests {
         P2pNetworkKadEntry {
             key,
             peer_id,
-            addrs: vec![],
+            addrs: vec![multiaddr!(Ip4([0; 4]), Tcp(1000_u16))],
             connection: super::ConnectionType::Connected,
         }
     }

p2p/tests/identify.rs

@@ -1,15 +1,31 @@
-use std::{collections::BTreeSet, time::Duration};
+use std::{
+    collections::{BTreeSet, HashSet},
+    time::Duration,
+};
 
-use multiaddr::Multiaddr;
-use p2p::PeerId;
+use multiaddr::{multiaddr, Multiaddr};
+use p2p::{
+    identity::SecretKey,
+    network::identify::{
+        stream::P2pNetworkIdentifyStreamState, P2pNetworkIdentify, P2pNetworkIdentifyAction,
+        P2pNetworkIdentifyStreamAction,
+    },
+    p2p_effects, p2p_timeout_effects,
+    token::{self, DiscoveryAlgorithm},
+    Data, P2pAction, P2pNetworkAction, P2pNetworkYamuxAction, PeerId,
+};
 use p2p_testing::{
-    cluster::{Cluster, ClusterBuilder, ClusterEvent},
-    event::RustNodeEvent,
+    cluster::{Cluster, ClusterBuilder, ClusterEvent, Listener},
+    event::{event_mapper_effect, RustNodeEvent},
     futures::TryStreamExt,
     predicates::{async_fn, listener_is_ready, peer_is_connected},
+    redux::{Action, State},
     rust_node::{RustNodeConfig, RustNodeId},
+    service::ClusterService,
     stream::ClusterStreamExt,
+    test_node::TestNode,
 };
+use redux::{ActionWithMeta, Store};
 
 #[tokio::test]
 async fn rust_node_to_rust_node() -> anyhow::Result<()> {
@@ -85,6 +101,171 @@ async fn rust_node_to_rust_node() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+/// Test that even if bad node spams many different listen_addrs we don't end up with duplicates
+async fn test_bad_node() -> anyhow::Result<()> {
+    let mut cluster = ClusterBuilder::new()
+        .ports_with_len(100)
+        .idle_duration(Duration::from_millis(100))
+        .start()
+        .await?;
+
+    let bad_node = cluster.add_rust_node(
+        RustNodeConfig::default()
+            .with_discovery(true)
+            .with_override(bad_node_effects),
+    )?;
+    let bad_node_peer_id = cluster.rust_node(bad_node).peer_id();
+    let bad_node_port = cluster.rust_node(bad_node).libp2p_port();
+
+    let node = cluster.add_rust_node(
+        RustNodeConfig::default()
+            .with_discovery(true)
+            .with_initial_peers([Listener::Rust(bad_node)]),
+    )?;
+
+    let mut not_identified = BTreeSet::from_iter([(node, bad_node_peer_id)]);
+    wait_for_identify(&mut cluster, &mut not_identified, Duration::from_secs(10)).await?;
+
+    let routing_table = &cluster
+        .rust_node(node)
+        .state()
+        .network
+        .scheduler
+        .discovery_state()
+        .expect("State must be initialized")
+        .routing_table;
+
+    let bad_peer_entry = routing_table
+        .look_up(&bad_node_peer_id.into())
+        .expect("Node not found");
+
+    let bad_peer_addresses = bad_peer_entry
+        .addrs
+        .iter()
+        .map(Clone::clone)
+        .collect::<HashSet<_>>();
+
+    let expected_addrs = [
+        multiaddr!(Ip4([127, 0, 0, 1]), Tcp(bad_node_port)),
+        multiaddr!(Ip4([127, 0, 0, 1]), Tcp(10500u16)),
+        multiaddr!(Ip6([0; 16]), Tcp(10500u16)),
+        multiaddr!(Ip6([1; 16]), Tcp(10500u16)),
+    ]
+    .into_iter()
+    .collect::<HashSet<_>>();
+
+    assert_eq!(bad_peer_addresses, expected_addrs);
+
+    Ok(())
+}
+
+fn bad_node_effects(
+    store: &mut Store<State, ClusterService, Action>,
+    action: ActionWithMeta<Action>,
+) {
+    {
+        let (action, meta) = action.split();
+        match action {
+            Action::P2p(a) => {
+                match a.clone() {
+                    P2pAction::Network(P2pNetworkAction::Identify(
+                        P2pNetworkIdentifyAction::Stream(P2pNetworkIdentifyStreamAction::New {
+                            addr,
+                            peer_id,
+                            stream_id,
+                            ..
+                        }),
+                    )) => {
+                        let state = store
+                            .state()
+                            .state()
+                            .network
+                            .scheduler
+                            .identify_state
+                            .find_identify_stream_state(&peer_id, &stream_id)
+                            .expect("Unable to find identify stream");
+
+                        if let P2pNetworkIdentifyStreamState::SendIdentify = state {
+                            let listen_addrs = vec![
+                                multiaddr!(Ip4([127, 0, 0, 1]), Tcp(10500u16)),
+                                multiaddr!(Ip4([127, 0, 0, 1]), Tcp(10500u16)),
+                                multiaddr!(Ip6([0; 16]), Tcp(10500u16)),
+                                multiaddr!(Ip6([0; 16]), Tcp(10500u16)),
+                                multiaddr!(Ip6([1; 16]), Tcp(10500u16)),
+                                multiaddr!(Ip6([1; 16]), Tcp(10500u16)),
+                                multiaddr!(Dns("domain.com"), Tcp(10530u16)),
+                                multiaddr!(Dns("domain.com"), Tcp(10530u16)),
+                                multiaddr!(Dns("domain.com"), Tcp(10530u16)),
+                                multiaddr!(Dns("domain.com"), Tcp(10530u16)),
+                                multiaddr!(Unix("domain.com")),
+                                multiaddr!(Https),
+                            ];
+
+                            let public_key = Some(SecretKey::rand().public_key());
+
+                            let protocols = vec![
+                                token::StreamKind::Identify(
+                                    token::IdentifyAlgorithm::Identify1_0_0,
+                                ),
+                                token::StreamKind::Broadcast(
+                                    p2p::token::BroadcastAlgorithm::Meshsub1_1_0,
+                                ),
+                                p2p::token::StreamKind::Rpc(token::RpcAlgorithm::Rpc0_0_1),
+                                p2p::token::StreamKind::Discovery(
+                                    DiscoveryAlgorithm::Kademlia1_0_0,
+                                ),
+                            ];
+
+                            let identify_msg = P2pNetworkIdentify {
+                                protocol_version: Some("ipfs/0.1.0".to_string()),
+                                agent_version: Some("openmina".to_owned()),
+                                public_key,
+                                listen_addrs,
+                                observed_addr: None,
+                                protocols,
+                            };
+
+                            let mut out = Vec::new();
+                            let identify_msg_proto = identify_msg.to_proto_message();
+
+                            prost::Message::encode_length_delimited(&identify_msg_proto, &mut out)
+                                .expect("Error converting message");
+
+                            store.dispatch(Action::P2p(
+                                P2pNetworkYamuxAction::OutgoingData {
+                                    addr,
+                                    stream_id,
+                                    data: Data(out.into_boxed_slice()),
+                                    flags: Default::default(),
+                                }
+                                .into(),
+                            ));
+
+                            store.dispatch(Action::P2p(
+                                P2pNetworkIdentifyStreamAction::Close {
+                                    addr,
+                                    peer_id,
+                                    stream_id,
+                                }
+                                .into(),
+                            ));
+                        }
+                    }
+
+                    a => {
+                        p2p_effects(store, meta.with_action(a.clone()));
+                    }
+                }
+                event_mapper_effect(store, a);
+            }
+            Action::Idle(_) => {
+                p2p_timeout_effects(store, &meta);
+            }
+        };
+    }
+}
+
 async fn wait_for_identify(
     cluster: &mut Cluster,
     nodes_peers: &mut BTreeSet<(RustNodeId, PeerId)>,