diff --git a/Cargo.lock b/Cargo.lock index c7c8049ba..2fc30e157 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -153,6 +153,17 @@ dependencies = [ "syn", ] +[[package]] +name = "async_io_stream" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6d7b9decdf35d8908a7e3ef02f64c5e9b1695e230154c0e8de3969142d9b94c" +dependencies = [ + "futures", + "pharos", + "rustc_version 0.4.0", +] + [[package]] name = "atomic-polyfill" version = "0.1.10" @@ -1113,6 +1124,15 @@ dependencies = [ "void", ] +[[package]] +name = "encoding_rs" +version = "0.8.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9852635589dc9f9ea1b6fe9f05b50ef208c85c834a562f0c6abb1c475736ec2b" +dependencies = [ + "cfg-if", +] + [[package]] name = "endian-type" version = "0.1.2" @@ -1139,6 +1159,25 @@ dependencies = [ "zeroize", ] +[[package]] +name = "enr" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "492a7e5fc2504d5fdce8e124d3e263b244a68b283cac67a69eda0cd43e0aebad" +dependencies = [ + "base64", + "bs58", + "bytes", + "hex", + "k256", + "log", + "rand 0.8.5", + "rlp", + "serde", + "sha3", + "zeroize", +] + [[package]] name = "enr" version = "0.7.0" @@ -1235,7 +1274,7 @@ dependencies = [ [[package]] name = "ethers-core" version = "1.0.2" -source = "git+https://github.com/gakonst/ethers-rs#a88d2d03e3340a4f9dd39fae9dfe094750db13d5" +source = "git+https://github.com/gakonst/ethers-rs#ea10f4b4dff471c1611eef45ce035524a9a7b550" dependencies = [ "arrayvec", "bytes", @@ -1257,6 +1296,41 @@ dependencies = [ "unicode-xid", ] +[[package]] +name = "ethers-providers" +version = "1.0.2" +source = "git+https://github.com/gakonst/ethers-rs#ea10f4b4dff471c1611eef45ce035524a9a7b550" +dependencies = [ + "async-trait", + "auto_impl", + "base64", + "enr 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)", + "ethers-core", + "futures-core", + "futures-timer", + "futures-util", + "getrandom 0.2.8", + "hashers", + "hex", + "http", + "once_cell", + "parking_lot 0.11.2", + "pin-project", + "reqwest", + "serde", + "serde_json", + "thiserror", + "tokio", + "tracing", + "tracing-futures", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "wasm-timer", + "web-sys", + "ws_stream_wasm", +] + [[package]] name = "ethnum" version = "1.3.0" @@ -1424,7 +1498,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e64b03909df88034c26dc1547e8970b91f98bdb65165d6a4e9110d94263dbb2c" dependencies = [ "gloo-timers", - "send_wrapper", + "send_wrapper 0.4.0", ] [[package]] @@ -1445,6 +1519,15 @@ dependencies = [ "slab", ] +[[package]] +name = "fxhash" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c31b6d751ae2c7f11320402d34e41349dd1016f8d5d45e48c4312bc8625af50c" +dependencies = [ + "byteorder", +] + [[package]] name = "generic-array" version = "0.14.6" @@ -1473,8 +1556,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c05aeb6a22b8f62540c194aac980f2115af067bfe15a0734d7277a768d396b31" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi 0.11.0+wasi-snapshot-preview1", + "wasm-bindgen", ] [[package]] @@ -1629,6 +1714,15 @@ dependencies = [ "ahash 0.8.2", ] +[[package]] +name = "hashers" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2bca93b15ea5a746f220e56587f71e73c6165eab783df9e26590069953e3c30" +dependencies = [ + "fxhash", +] + [[package]] name = "hashlink" version = "0.7.0" @@ -1904,6 +1998,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7a5bbe824c507c5da5956355e86a746d82e0e1464f65d862cc5e71da70e94b2c" dependencies = [ "cfg-if", + "js-sys", + "wasm-bindgen", + "web-sys", ] [[package]] @@ -2267,6 +2364,12 @@ dependencies = [ "syn", ] +[[package]] +name = "mime" +version = "0.3.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d" + [[package]] name = "minimal-lexical" version = "0.2.1" @@ -2657,6 +2760,16 @@ dependencies = [ "ucd-trie", ] +[[package]] +name = "pharos" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9567389417feee6ce15dd6527a8a1ecac205ef62c2932bcf3d9f6fc5b78b414" +dependencies = [ + "futures", + "rustc_version 0.4.0", +] + [[package]] name = "pin-project" version = "1.0.12" @@ -3033,6 +3146,40 @@ dependencies = [ "winapi", ] +[[package]] +name = "reqwest" +version = "0.11.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68cc60575865c7831548863cc02356512e3f1dc2f3f82cb837d7fc4cc8f3c97c" +dependencies = [ + "base64", + "bytes", + "encoding_rs", + "futures-core", + "futures-util", + "h2", + "http", + "http-body", + "hyper", + "ipnet", + "js-sys", + "log", + "mime", + "once_cell", + "percent-encoding", + "pin-project-lite", + "serde", + "serde_json", + "serde_urlencoded", + "tokio", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "winreg", +] + [[package]] name = "reth" version = "0.1.0" @@ -3335,8 +3482,12 @@ dependencies = [ "async-trait", "bytes", "either", + "enr 0.7.0 (git+https://github.com/sigp/enr)", + "ethers-core", + "ethers-providers", "fnv", "futures", + "hex", "linked_hash_set", "parking_lot 0.12.1", "pin-project", @@ -3353,6 +3504,7 @@ dependencies = [ "reth-tracing", "reth-transaction-pool", "secp256k1", + "tempfile", "thiserror", "tokio", "tokio-stream", @@ -3363,7 +3515,7 @@ dependencies = [ name = "reth-p2p" version = "0.1.0" dependencies = [ - "enr 0.7.0", + "enr 0.7.0 (git+https://github.com/sigp/enr)", "secp256k1", "serde", "serde_derive", @@ -3911,6 +4063,12 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f638d531eccd6e23b980caf34876660d38e265409d8e99b397ab71eb3612fad0" +[[package]] +name = "send_wrapper" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "930c0acf610d3fdb5e2ab6213019aaa04e227ebe9547b0649ba599b16d788bd7" + [[package]] name = "serde" version = "1.0.147" @@ -3942,6 +4100,18 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "serial_test" version = "0.9.0" @@ -4945,6 +5115,21 @@ version = "0.2.83" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c38c045535d93ec4f0b4defec448e4291638ee608530863b1e2ba115d4fff7f" +[[package]] +name = "wasm-timer" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be0ecb0db480561e9a7642b5d3e4187c128914e58aa84330b9493e3eb68c5e7f" +dependencies = [ + "futures", + "js-sys", + "parking_lot 0.11.2", + "pin-utils", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "web-sys" version = "0.3.60" @@ -5105,6 +5290,33 @@ version = "0.42.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f40009d85759725a34da6d89a94e63d7bdc50a862acf0dbc7c8e488f1edcb6f5" +[[package]] +name = "winreg" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "80d0f4e272c85def139476380b12f9ac60926689dd2e01d4923222f40580869d" +dependencies = [ + "winapi", +] + +[[package]] +name = "ws_stream_wasm" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47ca1ab42f5afed7fc332b22b6e932ca5414b209465412c8cdf0ad23bc0de645" +dependencies = [ + "async_io_stream", + "futures", + "js-sys", + "pharos", + "rustc_version 0.4.0", + "send_wrapper 0.5.0", + "thiserror", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wyz" version = "0.5.0" diff --git a/crates/net/network/Cargo.toml b/crates/net/network/Cargo.toml index 9ca90571d..ce674dc3c 100644 --- a/crates/net/network/Cargo.toml +++ b/crates/net/network/Cargo.toml @@ -50,5 +50,10 @@ secp256k1 = { version = "0.24", features = [ reth-interfaces = { path = "../../interfaces", features = ["test-utils"] } reth-provider = { path = "../../storage/provider", features = ["test-utils"] } reth-tracing = { path = "../../tracing" } +ethers-core = { git = "https://github.com/gakonst/ethers-rs", default-features = false } +ethers-providers = { git = "https://github.com/gakonst/ethers-rs", default-features = false } +enr = { git = "https://github.com/sigp/enr", features = ["serde", "rust-secp256k1"] } +hex = "0.4" +tempfile = "3.3" diff --git a/crates/net/network/src/config.rs b/crates/net/network/src/config.rs index a0d575312..ee9c762cc 100644 --- a/crates/net/network/src/config.rs +++ b/crates/net/network/src/config.rs @@ -143,6 +143,12 @@ impl NetworkConfigBuilder { } } + /// set a custom peer config for how peers are handled + pub fn peer_config(mut self, config: PeersConfig) -> Self { + self.peers_config = Some(config); + self + } + /// Sets the executor to use for spawning tasks. pub fn executor(mut self, executor: TaskExecutor) -> Self { self.executor = Some(executor); diff --git a/crates/net/network/src/lib.rs b/crates/net/network/src/lib.rs index 66214ba43..9f262c551 100644 --- a/crates/net/network/src/lib.rs +++ b/crates/net/network/src/lib.rs @@ -70,4 +70,4 @@ pub use config::NetworkConfig; pub use fetch::FetchClient; pub use manager::{NetworkEvent, NetworkManager}; pub use network::NetworkHandle; -pub use peers::PeersConfig; +pub use peers::{BanList, PeersConfig}; diff --git a/crates/net/network/src/peers/manager.rs b/crates/net/network/src/peers/manager.rs index 76b127192..5ad4187ff 100644 --- a/crates/net/network/src/peers/manager.rs +++ b/crates/net/network/src/peers/manager.rs @@ -3,11 +3,13 @@ use futures::StreamExt; use reth_eth_wire::DisconnectReason; use reth_primitives::PeerId; use std::{ - collections::{hash_map::Entry, HashMap, VecDeque}, - net::SocketAddr, + collections::{hash_map::Entry, HashMap, HashSet, VecDeque}, + fmt::Display, + net::{IpAddr, SocketAddr}, task::{Context, Poll}, time::Duration, }; +use thiserror::Error; use tokio::{ sync::mpsc, time::{Instant, Interval}, @@ -65,12 +67,15 @@ pub(crate) struct PeersManager { reputation_weights: ReputationChangeWeights, /// Tracks current slot stats. connection_info: ConnectionInfo, + /// Tracks unwanted ips/peer ids, + ban_list: BanList, } impl PeersManager { /// Create a new instance with the given config pub(crate) fn new(config: PeersConfig) -> Self { - let PeersConfig { refill_slots_interval, connection_info, reputation_weights } = config; + let PeersConfig { refill_slots_interval, connection_info, reputation_weights, ban_list } = + config; let (manager_tx, handle_rx) = mpsc::unbounded_channel(); Self { peers: Default::default(), @@ -83,6 +88,7 @@ impl PeersManager { refill_slots_interval, ), connection_info, + ban_list, } } @@ -93,11 +99,17 @@ impl PeersManager { /// Invoked when a new _incoming_ tcp connection is accepted. /// - /// Returns an Err with the configured limit with the number of accepted incoming connections - /// exceeded the configured limit. - pub(crate) fn on_inbound_pending_session(&mut self) -> Result<(), usize> { + /// returns an error if the inbound ip address is on the ban list or + /// we have reached our limit for max inbound connections + pub(crate) fn on_inbound_pending_session( + &mut self, + addr: IpAddr, + ) -> Result<(), InboundConnectionError> { + if self.ban_list.banned_ip(&addr) { + return Err(InboundConnectionError::IpBanned) + } if !self.connection_info.has_in_capacity() { - return Err(self.connection_info.max_inbound) + return Err(InboundConnectionError::ExceedsLimit(self.connection_info.max_inbound)) } // keep track of new connection @@ -117,6 +129,13 @@ impl PeersManager { /// If the reputation of the peer is below the `BANNED_REPUTATION` threshold, a disconnect will /// be scheduled. pub(crate) fn on_active_inbound_session(&mut self, peer_id: PeerId, addr: SocketAddr) { + // we only need to check the peer id here as the ip address will have been checked at + // on_inbound_pending_session + if self.ban_list.banned_peer(&peer_id) { + self.queued_actions.push_back(PeerAction::DisconnectBannedIncoming { peer_id }); + return + } + match self.peers.entry(peer_id) { Entry::Occupied(mut entry) => { let value = entry.get_mut(); @@ -175,9 +194,13 @@ impl PeersManager { /// Called for a newly discovered peer. /// - /// If the peer already exists, then the address will e updated. If the addresses differ, the + /// If the peer already exists, then the address will be updated. If the addresses differ, the /// old address is returned pub(crate) fn add_discovered_node(&mut self, peer_id: PeerId, addr: SocketAddr) { + if self.ban_list.is_banned(&addr.ip(), &peer_id) { + return + } + match self.peers.entry(peer_id) { Entry::Occupied(mut entry) => { let node = entry.get_mut(); @@ -429,7 +452,7 @@ pub enum PeerAction { /// Disconnect an existing connection. Disconnect { peer_id: PeerId, reason: Option }, /// Disconnect an existing incoming connection, because the peers reputation is below the - /// banned threshold. + /// banned threshold or is on the [`BanList`] DisconnectBannedIncoming { /// Peer id of the established connection. peer_id: PeerId, @@ -445,6 +468,8 @@ pub struct PeersConfig { pub connection_info: ConnectionInfo, /// How to weigh reputation changes pub reputation_weights: ReputationChangeWeights, + /// Restrictions on PeerIds and Ips + pub ban_list: BanList, } impl Default for PeersConfig { @@ -458,6 +483,122 @@ impl Default for PeersConfig { max_inbound: 30, }, reputation_weights: Default::default(), + ban_list: Default::default(), } } } + +impl PeersConfig { + /// A set of peer_ids and ip addr that we want to never connect to + pub fn with_ban_list(mut self, ban_list: BanList) -> Self { + self.ban_list = ban_list; + self + } +} + +/// Configuration for the automatic removal of unwanted peers +#[derive(Debug, Default)] +pub struct BanList { + /// banned peer ids + banned_peers: HashSet, + /// banned ips + banned_ips: HashSet, +} + +impl BanList { + /// creates a new instance from existing values + pub fn new(banned_peers: HashSet, banned_ips: HashSet) -> Self { + Self { banned_ips, banned_peers } + } + + /// checks the ban list to see if it contains the given ip + #[inline] + pub(self) fn banned_ip(&self, ip: &IpAddr) -> bool { + self.banned_ips.contains(ip) + } + + /// checks the banned list to see if it contains the given peer id + #[inline] + pub(self) fn banned_peer(&self, peer_id: &PeerId) -> bool { + self.banned_peers.contains(peer_id) + } + /// checks the ban list for the given ip address and peer_id and returns true + /// if the address is in the ban list + #[inline] + pub(self) fn is_banned(&self, ip: &IpAddr, peer_id: &PeerId) -> bool { + self.banned_ip(ip) || self.banned_peer(peer_id) + } +} + +#[derive(Debug, Error)] +pub enum InboundConnectionError { + ExceedsLimit(usize), + IpBanned, +} + +impl Display for InboundConnectionError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{self:?}") + } +} + +#[cfg(test)] +mod test { + use std::{ + collections::HashSet, + net::{IpAddr, Ipv4Addr, SocketAddr}, + }; + + use reth_primitives::{PeerId, H512}; + + use crate::{peers::PeerAction, BanList, PeersConfig}; + + use super::PeersManager; + + #[tokio::test] + async fn test_discovery_ban_list() { + let ip = IpAddr::V4(Ipv4Addr::new(127, 0, 1, 2)); + let socket_addr = SocketAddr::new(ip, 8008); + let ban_list = BanList::new(HashSet::default(), HashSet::from_iter(vec![ip])); + let config = PeersConfig::default().with_ban_list(ban_list); + let mut peer_manager = PeersManager::new(config); + peer_manager.add_discovered_node(H512::default(), socket_addr); + + assert!(peer_manager.peers.is_empty()); + } + + #[tokio::test] + async fn test_on_pending_ban_list() { + let ip = IpAddr::V4(Ipv4Addr::new(127, 0, 1, 2)); + let socket_addr = SocketAddr::new(ip, 8008); + let ban_list = BanList::new(HashSet::default(), HashSet::from_iter(vec![ip])); + let config = PeersConfig::default().with_ban_list(ban_list); + let mut peer_manager = PeersManager::new(config); + let a = peer_manager.on_inbound_pending_session(socket_addr.ip()); + // because we have no active peers this should be fine for testings + match a { + Ok(_) => panic!(), + Err(err) => match err { + super::InboundConnectionError::IpBanned {} => {} + super::InboundConnectionError::ExceedsLimit { .. } => { + panic!() + } + }, + } + } + + #[tokio::test] + async fn test_on_active_inbound_ban_list() { + let ip = IpAddr::V4(Ipv4Addr::new(127, 0, 1, 2)); + let socket_addr = SocketAddr::new(ip, 8008); + let given_peer_id: PeerId = H512::from_low_u64_ne(123403423412); + let ban_list = BanList::new(HashSet::from_iter(vec![given_peer_id]), HashSet::default()); + let config = PeersConfig::default().with_ban_list(ban_list); + let mut peer_manager = PeersManager::new(config); + peer_manager.on_active_inbound_session(given_peer_id, socket_addr); + + let Some(PeerAction::DisconnectBannedIncoming { peer_id }) = peer_manager.queued_actions.pop_front() else { panic!() }; + + assert_eq!(peer_id, given_peer_id) + } +} diff --git a/crates/net/network/src/peers/mod.rs b/crates/net/network/src/peers/mod.rs index 9ddf0a8c0..c2f918539 100644 --- a/crates/net/network/src/peers/mod.rs +++ b/crates/net/network/src/peers/mod.rs @@ -3,6 +3,6 @@ mod manager; mod reputation; -pub(crate) use manager::{PeerAction, PeersManager}; -pub use manager::{PeersConfig, PeersHandle}; +pub use manager::{BanList, PeersConfig, PeersHandle}; +pub(crate) use manager::{InboundConnectionError, PeerAction, PeersManager}; pub use reputation::{ReputationChangeKind, ReputationChangeWeights}; diff --git a/crates/net/network/src/session/handle.rs b/crates/net/network/src/session/handle.rs index 1f2afcbaa..9becad183 100644 --- a/crates/net/network/src/session/handle.rs +++ b/crates/net/network/src/session/handle.rs @@ -84,6 +84,7 @@ pub(crate) enum PendingSessionEvent { direction: Direction, error: Option, }, + /// Thrown when unable to establish a [`TcpStream`]. OutgoingConnectionError { remote_addr: SocketAddr, diff --git a/crates/net/network/src/session/mod.rs b/crates/net/network/src/session/mod.rs index 0083300ed..9942a5e0f 100644 --- a/crates/net/network/src/session/mod.rs +++ b/crates/net/network/src/session/mod.rs @@ -628,8 +628,8 @@ async fn authenticate( } } }; - let unauthed = UnauthedP2PStream::new(stream); + let auth = authenticate_stream( unauthed, session_id, @@ -697,7 +697,6 @@ async fn authenticate_stream( } } }; - PendingSessionEvent::Established { session_id, remote_addr, diff --git a/crates/net/network/src/swarm.rs b/crates/net/network/src/swarm.rs index ff13e55e1..320f22cfd 100644 --- a/crates/net/network/src/swarm.rs +++ b/crates/net/network/src/swarm.rs @@ -2,6 +2,7 @@ use crate::{ fetch::StatusUpdate, listener::{ConnectionListener, ListenerEvent}, message::{PeerMessage, PeerRequestSender}, + peers::InboundConnectionError, session::{Direction, SessionEvent, SessionId, SessionManager}, state::{NetworkState, StateAction}, }; @@ -172,10 +173,20 @@ where return Some(SwarmEvent::TcpListenerClosed { remote_addr: address }) } ListenerEvent::Incoming { stream, remote_addr } => { - if let Err(limit) = self.state_mut().peers_mut().on_inbound_pending_session() { - // We currently don't have additional capacity to establish a session from an - // incoming connection, so we drop the connection. - trace!(target: "net", %limit, ?remote_addr, "Exceeded incoming connection limit; dropping connection"); + if let Err(err) = + self.state_mut().peers_mut().on_inbound_pending_session(remote_addr.ip()) + { + match err { + InboundConnectionError::IpBanned => { + trace!(target: "net", ?remote_addr, "The incoming ip address is in the ban list"); + } + InboundConnectionError::ExceedsLimit(limit) => { + // We currently don't have additional capacity to establish a session + // from an incoming connection, so we drop + // the connection. + trace!(target: "net", %limit, ?remote_addr, "Exceeded incoming connection limit; dropping connection"); + } + } return None } diff --git a/crates/net/network/tests/it/connect.rs b/crates/net/network/tests/it/connect.rs index ad4b1ed99..9e8b9850f 100644 --- a/crates/net/network/tests/it/connect.rs +++ b/crates/net/network/tests/it/connect.rs @@ -1,12 +1,18 @@ //! Connection tests +use crate::NetworkEventStream; + use super::testnet::Testnet; +use enr::EnrPublicKey; +use ethers_core::utils::Geth; +use ethers_providers::{Http, Middleware, Provider}; use futures::StreamExt; use reth_discv4::{bootnodes::mainnet_nodes, Discv4Config}; -use reth_network::{NetworkConfig, NetworkEvent, NetworkManager}; +use reth_network::{BanList, NetworkConfig, NetworkEvent, NetworkManager, PeersConfig}; +use reth_primitives::PeerId; use reth_provider::test_utils::TestApi; use secp256k1::SecretKey; -use std::{collections::HashSet, sync::Arc}; +use std::{collections::HashSet, net::SocketAddr, sync::Arc}; #[tokio::test(flavor = "multi_thread")] async fn test_establish_connections() { @@ -80,3 +86,50 @@ async fn test_connect_with_boot_nodes() { dbg!(ev); } } + +#[tokio::test(flavor = "multi_thread")] +async fn test_incoming_node_id_blacklist() { + reth_tracing::init_tracing(); + let secret_key = SecretKey::new(&mut rand::thread_rng()); + + let reth_socket = SocketAddr::new([127, 0, 0, 1].into(), 30305); + + // instantiate geth and add ourselves as a peer + let temp_dir = tempfile::tempdir().unwrap().into_path(); + let geth = Geth::new().data_dir(temp_dir).disable_discovery().spawn(); + let geth_endpoint = SocketAddr::new([127, 0, 0, 1].into(), geth.port()); + let provider = Provider::::try_from(format!("http://{geth_endpoint}")).unwrap(); + + // get the peer id we should be expecting + let geth_peer_id: PeerId = + provider.node_info().await.unwrap().enr.public_key().encode_uncompressed().into(); + + let ban_list = BanList::new(HashSet::from_iter(vec![geth_peer_id]), HashSet::default()); + let peer_config = PeersConfig::default().with_ban_list(ban_list); + + let config = NetworkConfig::builder(Arc::new(TestApi::default()), secret_key) + .listener_addr(reth_socket) + .peer_config(peer_config) + .build(); + + let network = NetworkManager::new(config).await.unwrap(); + + let handle = network.handle().clone(); + tokio::task::spawn(network); + + // make geth connect to us + let our_peer_id = handle.peer_id(); + let our_enode = format!("enode://{}@{}", hex::encode(our_peer_id.0), reth_socket); + + provider.add_peer(our_enode).await.unwrap(); + + let events = handle.event_listener(); + let mut event_stream = NetworkEventStream::new(events); + + // check for session to be opened + let incoming_peer_id = event_stream.next_session_established().await.unwrap(); + assert_eq!(incoming_peer_id, geth_peer_id); + // check to see that the session was closed + let incoming_peer_id = event_stream.next_session_closed().await.unwrap(); + assert_eq!(incoming_peer_id, geth_peer_id); +} diff --git a/crates/net/network/tests/it/testnet.rs b/crates/net/network/tests/it/testnet.rs index fc21a8363..fed1698e9 100644 --- a/crates/net/network/tests/it/testnet.rs +++ b/crates/net/network/tests/it/testnet.rs @@ -289,6 +289,15 @@ impl NetworkEventStream { Self { inner } } + pub async fn next_session_closed(&mut self) -> Option { + while let Some(ev) = self.inner.next().await { + match ev { + NetworkEvent::SessionClosed { peer_id } => return Some(peer_id), + NetworkEvent::SessionEstablished { .. } => continue, + } + } + None + } /// Awaits the next event for an established session pub async fn next_session_established(&mut self) -> Option { while let Some(ev) = self.inner.next().await { diff --git a/deny.toml b/deny.toml index f9f99625e..ec621f7ea 100644 --- a/deny.toml +++ b/deny.toml @@ -43,7 +43,8 @@ allow = [ "BSD-3-Clause", "ISC", "Unicode-DFS-2016", - "OpenSSL" + "OpenSSL", + "Unlicense" ] # Allow 1 or more licenses on a per-crate basis, so that particular licenses