Repository: mintlayer/mintlayer-core
Pull request #144 (2022-04-22 10:14:00)

p2p: Enable timeouts for outbound connections
Previously, the SwarmManager could be DoS'ed by advertising an address
that didn't have an active listener and forcing the SwarmManager to
wait on that connection to succeeds.

Mock inteface's test might be a little unreliable because it's surprisingly hard to create a connection test that timeouts in localhost. If for some reason the code doesn't time out in CI/your computer/on Windows, I'll just disable it for now. It works on my computer so the actual timeout mechanisms works for both libp2p and mock.


Patch:
diff --git a/.github/workflows/loom.yml b/.github/workflows/loom.yml
index 11e09b7bb..8e5634430 100644
--- a/.github/workflows/loom.yml
+++ b/.github/workflows/loom.yml
@@ -28,4 +28,4 @@ jobs:
       with:
         submodules: recursive
     - name: Run permutation tests
-      run: cargo test --tests --release -p storage -p blockchain-storage
+      run: cargo test --tests --release --workspace
diff --git a/Cargo.lock b/Cargo.lock
index ab78046cf..380b5cc12 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2148,6 +2148,7 @@ dependencies = [
  "libp2p",
  "logging",
  "parity-scale-codec",
+ "portpicker",
  "rand 0.8.4",
  "test-utils",
  "tokio",
diff --git a/common/src/chain/block/block_v1.rs b/common/src/chain/block/block_v1.rs
index 282dc92ba..352df8d3d 100644
--- a/common/src/chain/block/block_v1.rs
+++ b/common/src/chain/block/block_v1.rs
@@ -6,8 +6,8 @@ use parity_scale_codec_derive::{Decode as DecodeDer, Encode as EncodeDer};
 #[derive(Debug, Clone, PartialEq, Eq, EncodeDer, DecodeDer)]
 pub struct BlockHeader {
     pub(super) hash_prev_block: Id<BlockV1>,
-    pub(super) tx_merkle_root: Option<H256>,
-    pub(super) witness_merkle_root: Option<H256>,
+    pub(super) tx_merkle_root: H256,
+    pub(super) witness_merkle_root: H256,
     pub(super) time: u32,
     pub(super) consensus_data: ConsensusData,
 }
@@ -22,11 +22,11 @@ impl BlockV1 {
     // This has to be the same its index in the Block enum
     pub const VERSION_BYTE: u8 = 0x01;
 
-    pub fn get_tx_merkle_root(&self) -> Option<H256> {
+    pub fn get_tx_merkle_root(&self) -> H256 {
         self.header.tx_merkle_root
     }
 
-    pub fn get_witness_merkle_root(&self) -> Option<H256> {
+    pub fn get_witness_merkle_root(&self) -> H256 {
         self.header.witness_merkle_root
     }
 
diff --git a/common/src/chain/block/mod.rs b/common/src/chain/block/mod.rs
index 42d3549e4..c77f42202 100644
--- a/common/src/chain/block/mod.rs
+++ b/common/src/chain/block/mod.rs
@@ -29,33 +29,27 @@ use parity_scale_codec::{Decode, Encode};
 
 pub fn calculate_tx_merkle_root(
     transactions: &[Transaction],
-) -> Result<Option<H256>, merkle::MerkleTreeFormError> {
-    if transactions.is_empty() {
-        return Ok(None);
-    }
+) -> Result<H256, merkle::MerkleTreeFormError> {
     if transactions.len() == 1 {
         // using bitcoin's way, blocks that only have the coinbase use their coinbase as the merkleroot
-        return Ok(Some(transactions[0].get_id().get()));
+        return Ok(transactions[0].get_id().get());
     }
     let hashes: Vec<H256> = transactions.iter().map(|tx| tx.get_id().get()).collect();
     let t = merkle::merkletree_from_vec(&hashes)?;
-    Ok(Some(t.root()))
+    Ok(t.root())
 }
 
 pub fn calculate_witness_merkle_root(
     transactions: &[Transaction],
-) -> Result<Option<H256>, merkle::MerkleTreeFormError> {
-    if transactions.is_empty() {
-        return Ok(None);
-    }
+) -> Result<H256, merkle::MerkleTreeFormError> {
     // TODO: provide implementation based on real serialization instead of get_id()
     if transactions.len() == 1 {
         // using bitcoin's way, blocks that only have the coinbase use their coinbase as the merkleroot
-        return Ok(Some(transactions[0].get_id().get()));
+        return Ok(transactions[0].get_id().get());
     }
     let hashes: Vec<H256> = transactions.iter().map(|tx| tx.get_id().get()).collect();
     let t = merkle::merkletree_from_vec(&hashes)?;
-    Ok(Some(t.root()))
+    Ok(t.root())
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -119,13 +113,13 @@ impl Block {
         }
     }
 
-    pub fn get_merkle_root(&self) -> Option<H256> {
+    pub fn get_merkle_root(&self) -> H256 {
         match &self {
             Block::V1(blk) => blk.get_tx_merkle_root(),
         }
     }
 
-    pub fn get_witness_merkle_root(&self) -> Option<H256> {
+    pub fn get_witness_merkle_root(&self) -> H256 {
         match &self {
             Block::V1(blk) => blk.get_witness_merkle_root(),
         }
@@ -166,7 +160,7 @@ impl Idable<Block> for Block {
 
 #[cfg(test)]
 mod tests {
-    use crate::chain::transaction::Transaction;
+    use crate::{chain::transaction::Transaction, primitives::merkle::MerkleTreeFormError};
 
     use super::*;
     use rand::Rng;
@@ -177,8 +171,8 @@ mod tests {
 
         let header = BlockHeader {
             consensus_data: ConsensusData::None,
-            tx_merkle_root: Some(H256::from_low_u64_be(rng.gen())),
-            witness_merkle_root: Some(H256::from_low_u64_be(rng.gen())),
+            tx_merkle_root: H256::from_low_u64_be(rng.gen()),
+            witness_merkle_root: H256::from_low_u64_be(rng.gen()),
             hash_prev_block: Id::new(&H256::zero()),
             time: rng.gen(),
         };
@@ -188,29 +182,28 @@ mod tests {
             transactions: Vec::new(),
         });
         let _res = calculate_tx_merkle_root(block.get_transactions());
-        assert_eq!(_res.unwrap(), None);
+        assert_eq!(_res.unwrap_err(), MerkleTreeFormError::TooSmall(0));
     }
 
     #[test]
-    fn block_merkleroot_only_one_transaction() {
+    fn block_merkleroot_only_coinbase() {
         let mut rng = rand::thread_rng();
 
         let header = BlockHeader {
             consensus_data: ConsensusData::None,
-            tx_merkle_root: Some(H256::from_low_u64_be(rng.gen())),
-            witness_merkle_root: Some(H256::from_low_u64_be(rng.gen())),
+            tx_merkle_root: H256::from_low_u64_be(rng.gen()),
+            witness_merkle_root: H256::from_low_u64_be(rng.gen()),
             hash_prev_block: Id::new(&H256::zero()),
             time: rng.gen(),
         };
 
-        let one_transaction = Transaction::new(0, Vec::new(), Vec::new(), 0).unwrap();
+        let coinbase = Transaction::new(0, Vec::new(), Vec::new(), 0).unwrap();
 
         let block = Block::V1(BlockV1 {
             header,
-            transactions: vec![one_transaction.clone()],
+            transactions: vec![coinbase.clone()],
         });
         let res = calculate_tx_merkle_root(block.get_transactions()).unwrap();
-        let res = res.unwrap();
-        assert_eq!(res, one_transaction.get_id().get());
+        assert_eq!(res, coinbase.get_id().get());
     }
 }
diff --git a/p2p/Cargo.toml b/p2p/Cargo.toml
index 85fad292c..18b81234a 100644
--- a/p2p/Cargo.toml
+++ b/p2p/Cargo.toml
@@ -26,13 +26,16 @@ features = ["floodsub", "mdns", "mplex", "noise", "streaming", "tcp-async-io"]
 
 [dependencies.tokio]
 version = "1"
-features = ["io-util", "macros", "net", "rt", "rt-multi-thread", "sync"]
+features = ["io-util", "macros", "net", "rt", "rt-multi-thread", "sync", "time"]
 
 [dependencies.parity-scale-codec]
 default-features = false
 features = ['derive']
 version = '2.0.0'
 
+[dev-dependencies]
+portpicker = "0.1.1"
+
 [dev-dependencies.test-utils]
 version = "0.1.0"
 path = "test-utils"
diff --git a/p2p/src/lib.rs b/p2p/src/lib.rs
index 89793b45b..b6854f6a5 100644
--- a/p2p/src/lib.rs
+++ b/p2p/src/lib.rs
@@ -14,6 +14,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+#![cfg(not(loom))]
 
 use crate::net::{ConnectivityService, FloodsubService, NetworkService};
 use common::chain::ChainConfig;
@@ -59,7 +60,7 @@ where
         addr: T::Address,
         config: Arc<ChainConfig>,
     ) -> error::Result<Self> {
-        let (conn, flood) = T::start(addr, &[], &[]).await?;
+        let (conn, flood) = T::start(addr, &[], &[], std::time::Duration::from_secs(10)).await?;
         let (tx_swarm, rx_swarm) = mpsc::channel(16);
         let (tx_sync, rx_sync) = mpsc::channel(16);
         let (tx_peer, rx_peer) = mpsc::channel(16);
diff --git a/p2p/src/net/libp2p/mod.rs b/p2p/src/net/libp2p/mod.rs
index bfefda649..ed78ca9c4 100644
--- a/p2p/src/net/libp2p/mod.rs
+++ b/p2p/src/net/libp2p/mod.rs
@@ -192,6 +192,7 @@ impl NetworkService for Libp2pService {
         addr: Self::Address,
         strategies: &[Self::Strategy],
         topics: &[FloodsubTopic],
+        timeout: std::time::Duration,
     ) -> error::Result<(Self::ConnectivityHandle, Self::FloodsubHandle)> {
         let id_keys = identity::Keypair::generate_ed25519();
         let peer_id = id_keys.public().to_peer_id();
@@ -203,6 +204,7 @@ impl NetworkService for Libp2pService {
             .upgrade(upgrade::Version::V1)
             .authenticate(noise::NoiseConfig::xx(noise_keys).into_authenticated())
             .multiplex(mplex::MplexConfig::new())
+            .outbound_timeout(timeout)
             .boxed();
 
         let swarm = {
@@ -464,6 +466,8 @@ impl SocketService for Libp2pSocket {
 mod tests {
     use super::*;
     use crate::net;
+    use std::time::Duration;
+    use tokio::net::TcpListener;
 
     #[derive(Debug, Encode, Decode, PartialEq, Eq, Copy, Clone)]
     struct Transaction {
@@ -473,7 +477,13 @@ mod tests {
 
     #[tokio::test]
     async fn test_connect_new() {
-        let service = Libp2pService::start("/ip6/::1/tcp/8900".parse().unwrap(), &[], &[]).await;
+        let service = Libp2pService::start(
+            "/ip6/::1/tcp/8900".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
         assert!(service.is_ok());
     }
 
@@ -481,10 +491,22 @@ mod tests {
     #[ignore]
     #[tokio::test]
     async fn test_connect_new_addrinuse() {
-        let service = Libp2pService::start("/ip6/::1/tcp/8901".parse().unwrap(), &[], &[]).await;
+        let service = Libp2pService::start(
+            "/ip6/::1/tcp/8901".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
         assert!(service.is_ok());
 
-        let service = Libp2pService::start("/ip6/::1/tcp/8901".parse().unwrap(), &[], &[]).await;
+        let service = Libp2pService::start(
+            "/ip6/::1/tcp/8901".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
 
         match service {
             Err(e) => {
@@ -498,8 +520,20 @@ mod tests {
     // and having `service2` trying to connect to `service1`
     #[tokio::test]
     async fn test_connect_accept() {
-        let service1 = Libp2pService::start("/ip6/::1/tcp/8902".parse().unwrap(), &[], &[]).await;
-        let service2 = Libp2pService::start("/ip6/::1/tcp/8903".parse().unwrap(), &[], &[]).await;
+        let service1 = Libp2pService::start(
+            "/ip6/::1/tcp/8902".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
+        let service2 = Libp2pService::start(
+            "/ip6/::1/tcp/8903".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
         assert!(service1.is_ok());
         assert!(service2.is_ok());
 
@@ -519,9 +553,14 @@ mod tests {
     #[tokio::test]
     async fn test_connect_peer_id_missing() {
         let addr: Multiaddr = "/ip6/::1/tcp/8904".parse().unwrap();
-        let (mut service, _) = Libp2pService::start("/ip6/::1/tcp/8905".parse().unwrap(), &[], &[])
-            .await
-            .unwrap();
+        let (mut service, _) = Libp2pService::start(
+            "/ip6/::1/tcp/8905".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await
+        .unwrap();
         match service.connect(addr).await {
             Ok(_) => panic!("connect succeeded without peer id"),
             Err(e) => {
@@ -540,8 +579,20 @@ mod tests {
     // it decodes to the same transaction that was sent.
     #[tokio::test]
     async fn test_peer_send() {
-        let service1 = Libp2pService::start("/ip6/::1/tcp/8905".parse().unwrap(), &[], &[]).await;
-        let service2 = Libp2pService::start("/ip6/::1/tcp/8906".parse().unwrap(), &[], &[]).await;
+        let service1 = Libp2pService::start(
+            "/ip6/::1/tcp/8905".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
+        let service2 = Libp2pService::start(
+            "/ip6/::1/tcp/8906".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
 
         let (mut service1, _) = service1.unwrap();
         let (mut service2, _) = service2.unwrap();
@@ -583,8 +634,20 @@ mod tests {
     // it decodes to the same transaction that was sent.
     #[tokio::test]
     async fn test_peer_recv() {
-        let service1 = Libp2pService::start("/ip6/::1/tcp/8907".parse().unwrap(), &[], &[]).await;
-        let service2 = Libp2pService::start("/ip6/::1/tcp/8908".parse().unwrap(), &[], &[]).await;
+        let service1 = Libp2pService::start(
+            "/ip6/::1/tcp/8907".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
+        let service2 = Libp2pService::start(
+            "/ip6/::1/tcp/8908".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
 
         let (mut service1, _) = service1.unwrap();
         let (mut service2, _) = service2.unwrap();
@@ -618,8 +681,20 @@ mod tests {
     // end and decoded as three separate transactions
     #[tokio::test]
     async fn test_peer_buffered_recv() {
-        let service1 = Libp2pService::start("/ip6/::1/tcp/8909".parse().unwrap(), &[], &[]).await;
-        let service2 = Libp2pService::start("/ip6/::1/tcp/8910".parse().unwrap(), &[], &[]).await;
+        let service1 = Libp2pService::start(
+            "/ip6/::1/tcp/8909".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
+        let service2 = Libp2pService::start(
+            "/ip6/::1/tcp/8910".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
 
         let (mut service1, _) = service1.unwrap();
         let (mut service2, _) = service2.unwrap();
@@ -655,8 +730,20 @@ mod tests {
     // that is too big and verify that it is rejected
     #[tokio::test]
     async fn test_too_large_message_size() {
-        let service1 = Libp2pService::start("/ip6/::1/tcp/8911".parse().unwrap(), &[], &[]).await;
-        let service2 = Libp2pService::start("/ip6/::1/tcp/8912".parse().unwrap(), &[], &[]).await;
+        let service1 = Libp2pService::start(
+            "/ip6/::1/tcp/8911".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
+        let service2 = Libp2pService::start(
+            "/ip6/::1/tcp/8912".parse().unwrap(),
+            &[],
+            &[],
+            Duration::from_secs(10),
+        )
+        .await;
 
         let (mut service1, _) = service1.unwrap();
         let (mut service2, _) = service2.unwrap();
@@ -835,4 +922,52 @@ mod tests {
             }]
         );
     }
+
+    impl PartialEq for Libp2pSocket {
+        fn eq(&self, _other: &Libp2pSocket) -> bool {
+            true
+        }
+    }
+
+    // try to connect to a service that is not listening with a small timeout and verify that the connection fails
+    // TODO: verify on windows/mac
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn test_connect_with_timeout() {
+        let (mut service, _) = Libp2pService::start(
+            test_utils::make_address("/ip6/::1/tcp/"),
+            &[],
+            &[],
+            Duration::from_secs(2),
+        )
+        .await
+        .unwrap();
+
+        let port = portpicker::pick_unused_port().unwrap();
+        let id: PeerId = "12D3KooWE3kBRAnn6jxZMdK1JMWx1iHtR1NKzXSRv5HLTmfD9u9c".parse().unwrap();
+        let mut addr: Multiaddr = format!("/ip6/::1/tcp/{}", port).parse().unwrap();
+        addr.push(Protocol::P2p(id.into()));
+
+        // first try to connect to address nobody is listening to
+        // and verify that the connection is refused immediately
+        let start = std::time::SystemTime::now();
+        assert_eq!(
+            service.connect(addr.clone()).await,
+            Err(P2pError::SocketError(std::io::ErrorKind::ConnectionRefused))
+        );
+        assert_eq!(
+            std::time::SystemTime::now().duration_since(start).unwrap().as_secs(),
+            0
+        );
+
+        // then create a socket that listens to the address and verify that it takes
+        // 2 seconds to get the `ConnectionRefused` error, as expected
+        let _service = TcpListener::bind(format!("[::1]:{}", port)).await.unwrap();
+        let start = std::time::SystemTime::now();
+        assert_eq!(
+            service.connect(addr).await,
+            Err(P2pError::SocketError(std::io::ErrorKind::ConnectionRefused))
+        );
+        assert!(std::time::SystemTime::now().duration_since(start).unwrap().as_secs() >= 2);
+    }
 }
diff --git a/p2p/src/net/mock/backend.rs b/p2p/src/net/mock/backend.rs
index 1d7203a5e..fac245a89 100644
--- a/p2p/src/net/mock/backend.rs
+++ b/p2p/src/net/mock/backend.rs
@@ -23,6 +23,7 @@ use crate::{
 };
 use async_trait::async_trait;
 use futures::FutureExt;
+// use futures_timer::Delay;
 use logging::log;
 use parity_scale_codec::{Decode, Encode};
 use std::{
@@ -51,6 +52,9 @@ pub struct Backend {
 
     /// TX channel for sending events to the frontend
     _flood_tx: mpsc::Sender<types::FloodsubEvent>,
+
+    /// Timeout for outbound operations
+    timeout: std::time::Duration,
 }
 
 impl Backend {
@@ -60,6 +64,7 @@ impl Backend {
         cmd_rx: mpsc::Receiver<types::Command>,
         conn_tx: mpsc::Sender<types::ConnectivityEvent>,
         _flood_tx: mpsc::Sender<types::FloodsubEvent>,
+        timeout: std::time::Duration,
     ) -> Self {
         Self {
             addr,
@@ -67,6 +72,7 @@ impl Backend {
             cmd_rx,
             conn_tx,
             _flood_tx,
+            timeout,
         }
     }
 
@@ -90,10 +96,17 @@ impl Backend {
                             continue;
                         }
 
-                        let _ = match TcpStream::connect(addr).await {
-                            Ok(socket) => response.send(Ok(socket)),
-                            Err(e) => response.send(Err(e.into())),
-                        };
+                        tokio::select! {
+                            _ = tokio::time::sleep(self.timeout) => {
+                                let _ = response.send(Err(
+                                    P2pError::SocketError(std::io::ErrorKind::ConnectionRefused))
+                                );
+                            }
+                            res = TcpStream::connect(addr) => match res {
+                                Ok(socket) => { let _ = response.send(Ok(socket)); },
+                                Err(e) => { let _ = response.send(Err(e.into())); },
+                            }
+                        }
                     }
                 }
             }
diff --git a/p2p/src/net/mock/mod.rs b/p2p/src/net/mock/mod.rs
index c6c9e21e1..0de3ced5c 100644
--- a/p2p/src/net/mock/mod.rs
+++ b/p2p/src/net/mock/mod.rs
@@ -98,6 +98,7 @@ impl NetworkService for MockService {
         addr: Self::Address,
         _strategies: &[Self::Strategy],
         _topics: &[FloodsubTopic],
+        timeout: std::time::Duration,
     ) -> error::Result<(Self::ConnectivityHandle, Self::FloodsubHandle)> {
         let (cmd_tx, cmd_rx) = mpsc::channel(16);
         let (conn_tx, conn_rx) = mpsc::channel(16);
@@ -105,7 +106,7 @@ impl NetworkService for MockService {
         let socket = TcpListener::bind(addr).await?;
 
         tokio::spawn(async move {
-            let mut mock = backend::Backend::new(addr, socket, cmd_rx, conn_tx, flood_tx);
+            let mut mock = backend::Backend::new(addr, socket, cmd_rx, conn_tx, flood_tx, timeout);
             let _ = mock.run().await;
         });
 
@@ -225,7 +226,7 @@ mod tests {
     use std::net::SocketAddr;
     use std::sync::Arc;
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    use tokio::net::TcpStream;
+    use tokio::net::{TcpStream, UdpSocket};
 
     #[derive(Debug, Encode, Decode, PartialEq, Eq)]
     struct Transaction {
@@ -235,19 +236,43 @@ mod tests {
 
     #[tokio::test]
     async fn test_new() {
-        let srv_ipv4 = MockService::start("127.0.0.1:5555".parse().unwrap(), &[], &[]).await;
+        let srv_ipv4 = MockService::start(
+            "127.0.0.1:5555".parse().unwrap(),
+            &[],
+            &[],
+            std::time::Duration::from_secs(10),
+        )
+        .await;
         assert!(srv_ipv4.is_ok());
 
         // address already in use
-        let err = MockService::start("127.0.0.1:5555".parse().unwrap(), &[], &[]).await;
+        let err = MockService::start(
+            "127.0.0.1:5555".parse().unwrap(),
+            &[],
+            &[],
+            std::time::Duration::from_secs(10),
+        )
+        .await;
         assert!(err.is_err());
 
         // bind to IPv6 localhost
-        let srv_ipv6 = MockService::start("[::1]:5555".parse().unwrap(), &[], &[]).await;
+        let srv_ipv6 = MockService::start(
+            "[::1]:5555".parse().unwrap(),
+            &[],
+            &[],
+            std::time::Duration::from_secs(10),
+        )
+        .await;
         assert!(srv_ipv6.is_ok());
 
         // address already in use
-        let s_ipv6 = MockService::start("[::1]:5555".parse().unwrap(), &[], &[]).await;
+        let s_ipv6 = MockService::start(
+            "[::1]:5555".parse().unwrap(),
+            &[],
+            &[],
+            std::time::Duration::from_secs(10),
+        )
+        .await;
         assert!(s_ipv6.is_err());
     }
 
@@ -266,7 +291,13 @@ mod tests {
         });
 
         // create service that is used for testing `connect()`
-        let srv = MockService::start("127.0.0.1:7777".parse().unwrap(), &[], &[]).await;
+        let srv = MockService::start(
+            "127.0.0.1:7777".parse().unwrap(),
+            &[],
+            &[],
+            std::time::Duration::from_secs(10),
+        )
+        .await;
         assert!(srv.is_ok());
         let (mut srv, _) = srv.unwrap();
 
@@ -289,8 +320,14 @@ mod tests {
     async fn test_accept() {
         // create service that is used for testing `accept()`
         let addr: SocketAddr = "[::1]:9999".parse().unwrap();
-        let (mut srv, _) =
-            MockService::start("[::1]:9999".parse().unwrap(), &[], &[]).await.unwrap();
+        let (mut srv, _) = MockService::start(
+            "[::1]:9999".parse().unwrap(),
+            &[],
+            &[],
+            std::time::Duration::from_secs(10),
+        )
+        .await
+        .unwrap();
 
         let (acc, con) = tokio::join!(srv.poll_next(), TcpStream::connect(addr));
         assert!(acc.is_ok());
@@ -303,7 +340,10 @@ mod tests {
     #[tokio::test]
     async fn test_peer_send() {
         let addr: SocketAddr = "[::1]:11112".parse().unwrap();
-        let (mut server, _) = MockService::start(addr, &[], &[]).await.unwrap();
+        let (mut server, _) =
+            MockService::start(addr, &[], &[], std::time::Duration::from_secs(10))
+                .await
+                .unwrap();
         let remote_fut = TcpStream::connect(addr);
 
         let (server_res, remote_res) = tokio::join!(server.poll_next(), remote_fut);
@@ -350,7 +390,10 @@ mod tests {
     async fn test_peer_recv() {
         // create a `MockService`, connect to it with a `TcpStream` and exchange data
         let addr: SocketAddr = "[::1]:11113".parse().unwrap();
-        let (mut server, _) = MockService::start(addr, &[], &[]).await.unwrap();
+        let (mut server, _) =
+            MockService::start(addr, &[], &[], std::time::Duration::from_secs(10))
+                .await
+                .unwrap();
         let remote_fut = TcpStream::connect(addr);
 
         let (server_res, remote_res) = tokio::join!(server.poll_next(), remote_fut);
@@ -391,4 +434,44 @@ mod tests {
         assert!(peer_res.is_ok());
         assert_eq!(peer_res.unwrap(), tx);
     }
+
+    impl PartialEq for MockSocket {
+        fn eq(&self, _other: &MockSocket) -> bool {
+            true
+        }
+    }
+
+    // try to connect to a service that is not listening with a small timeout
+    // and verify that the connection fails
+    // TODO: verify on windows/mac
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn test_connect_with_timeout() {
+        let addr: SocketAddr = test_utils::make_address("[::1]:");
+        let (mut service, _) =
+            MockService::start(addr, &[], &[], std::time::Duration::from_secs(2))
+                .await
+                .unwrap();
+
+        let port = portpicker::pick_unused_port().unwrap();
+        let conn_addr: SocketAddr = format!("[::1]:{}", port).parse().unwrap();
+        let start = std::time::SystemTime::now();
+        assert_eq!(
+            service.connect(conn_addr).await,
+            Err(P2pError::SocketError(std::io::ErrorKind::ConnectionRefused))
+        );
+        assert_eq!(
+            std::time::SystemTime::now().duration_since(start).unwrap().as_secs(),
+            0
+        );
+
+        // try to connect to a non-routable host
+        let conn_addr: SocketAddr = "10.255.255.255:8888".parse().unwrap();
+        let start = std::time::SystemTime::now();
+        assert_eq!(
+            service.connect(conn_addr).await,
+            Err(P2pError::SocketError(std::io::ErrorKind::ConnectionRefused))
+        );
+        assert!(std::time::SystemTime::now().duration_since(start).unwrap().as_secs() >= 2);
+    }
 }
diff --git a/p2p/src/net/mod.rs b/p2p/src/net/mod.rs
index 26b9353ea..d65017e4a 100644
--- a/p2p/src/net/mod.rs
+++ b/p2p/src/net/mod.rs
@@ -110,10 +110,12 @@ pub trait NetworkService {
     /// `addr` - socket address for incoming P2P traffic
     /// `strategies` - list of strategies that are used for peer discovery
     /// `topics` - list of floodsub topics that the implementation should subscribe to
+    /// `timeout` - timeout for outbound connections
     async fn start(
         addr: Self::Address,
         strategies: &[Self::Strategy],
         topics: &[FloodsubTopic],
+        timeout: std::time::Duration,
     ) -> error::Result<(Self::ConnectivityHandle, Self::FloodsubHandle)>;
 }
 
diff --git a/p2p/src/swarm.rs b/p2p/src/swarm.rs
index 02d8a10a0..a3e4f7a29 100644
--- a/p2p/src/swarm.rs
+++ b/p2p/src/swarm.rs
@@ -15,6 +15,7 @@
 //
 // Author(s): A. Altonen
 #![allow(unused, dead_code)]
+#![cfg(not(loom))]
 
 ///! Swarm manager
 ///!
@@ -413,7 +414,7 @@ mod tests {
         T::ConnectivityHandle: ConnectivityService<T>,
     {
         let config = Arc::new(config::create_mainnet());
-        let (conn, _) = T::start(addr, &[], &[]).await.unwrap();
+        let (conn, _) = T::start(addr, &[], &[], std::time::Duration::from_secs(10)).await.unwrap();
         let (_, rx) = tokio::sync::mpsc::channel(16);
         let (tx_sync, mut rx_sync) = tokio::sync::mpsc::channel(16);
         let (tx_peer, _) = tokio::sync::mpsc::channel(16);
diff --git a/p2p/src/sync/mod.rs b/p2p/src/sync/mod.rs
index 2cc42b701..eff08d13b 100644
--- a/p2p/src/sync/mod.rs
+++ b/p2p/src/sync/mod.rs
@@ -14,6 +14,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+#![cfg(not(loom))]
 #![allow(unused)]
 
 use crate::{
@@ -193,7 +194,8 @@ mod tests {
         T::FloodsubHandle: FloodsubService<T>,
     {
         let config = Arc::new(config::create_mainnet());
-        let (_, flood) = T::start(addr, &[], &[]).await.unwrap();
+        let (_, flood) =
+            T::start(addr, &[], &[], std::time::Duration::from_secs(10)).await.unwrap();
         let (tx_sync, rx_sync) = tokio::sync::mpsc::channel(16);
         let (tx_peer, rx_peer) = tokio::sync::mpsc::channel(16);
 
diff --git a/script/src/interpreter.rs b/script/src/interpreter.rs
index bdf6806c9..ee741123d 100644
--- a/script/src/interpreter.rs
+++ b/script/src/interpreter.rs
@@ -526,7 +526,7 @@ fn op_hash<T: AsRef<[u8]>>(stack: &mut Stack, f: impl FnOnce(&[u8]) -> T) -> cra
     Ok(())
 }
 
-#[cfg(test)]
+#[cfg(all(test, not(loom)))]
 mod test {
     use super::*;
     use crate::opcodes::all as opc;
diff --git a/script/src/lib.rs b/script/src/lib.rs
index 7f705d0d4..fece362f0 100644
--- a/script/src/lib.rs
+++ b/script/src/lib.rs
@@ -55,7 +55,7 @@ mod interpreter;
 pub mod opcodes;
 pub mod script;
 pub mod sighash;
-#[cfg(test)]
+#[cfg(all(test, not(loom)))]
 mod test;
 
 pub use crate::script::{Builder, Script};


Test patch:
diff --git a/p2p/test-utils/src/lib.rs b/p2p/test-utils/src/lib.rs
index 517daefa0..e78761ae5 100644
--- a/p2p/test-utils/src/lib.rs
+++ b/p2p/test-utils/src/lib.rs
@@ -14,6 +14,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+#![cfg(not(loom))]
 
 use common::chain::ChainConfig;
 use libp2p::Multiaddr;
@@ -64,7 +65,7 @@ pub async fn create_two_mock_peers(
     config: Arc<ChainConfig>,
 ) -> (Peer<MockService>, Peer<MockService>) {
     let addr: SocketAddr = make_address("[::1]:");
-    let (mut server, _) = MockService::start(addr, &[], &[]).await.unwrap();
+    let (mut server, _) = MockService::start(addr, &[], &[], std::time::Duration::from_secs(10)).await.unwrap();
     let peer_fut = TcpStream::connect(addr);
 
     let (remote_res, local_res) = tokio::join!(server.poll_next(), peer_fut);
@@ -116,10 +117,10 @@ pub async fn create_two_libp2p_peers(
     config: Arc<ChainConfig>,
 ) -> (Peer<Libp2pService>, Peer<Libp2pService>) {
     let addr1: Multiaddr = make_address("/ip6/::1/tcp/");
-    let (mut server1, _) = Libp2pService::start(addr1, &[], &[]).await.unwrap();
+    let (mut server1, _) = Libp2pService::start(addr1, &[], &[], std::time::Duration::from_secs(10)).await.unwrap();
 
     let addr2: Multiaddr = make_address("/ip6/::1/tcp/");
-    let (mut server2, _) = Libp2pService::start(addr2, &[], &[]).await.unwrap();
+    let (mut server2, _) = Libp2pService::start(addr2, &[], &[], std::time::Duration::from_secs(10)).await.unwrap();
 
     let server1_conn_fut = server1.connect(server2.local_addr().clone());
 
diff --git a/p2p/tests/connectivity.rs b/p2p/tests/connectivity.rs
index da5ad9e9b..2b33d8199 100644
--- a/p2p/tests/connectivity.rs
+++ b/p2p/tests/connectivity.rs
@@ -14,6 +14,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+#![cfg(not(loom))]
 
 use common::{
     chain::{config, ChainConfig},
diff --git a/p2p/tests/handshake.rs b/p2p/tests/handshake.rs
index a66cd1f17..1cbc7fb5a 100644
--- a/p2p/tests/handshake.rs
+++ b/p2p/tests/handshake.rs
@@ -14,6 +14,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+#![cfg(not(loom))]
 
 use common::{
     chain::config,
diff --git a/p2p/tests/libp2p.rs b/p2p/tests/libp2p.rs
index 1383cc5a3..2fe6907e2 100644
--- a/p2p/tests/libp2p.rs
+++ b/p2p/tests/libp2p.rs
@@ -14,6 +14,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+#![cfg(not(loom))]
 extern crate test_utils;
 
 use libp2p::{multiaddr::Protocol, Multiaddr};
@@ -32,14 +33,24 @@ use p2p::{
 #[tokio::test]
 async fn test_libp2p_peer_discovery() {
     let addr: Multiaddr = test_utils::make_address("/ip6/::1/tcp/");
-    let (mut serv, _) = Libp2pService::start(addr.clone(), &[Libp2pStrategy::MulticastDns], &[])
-        .await
-        .unwrap();
+    let (mut serv, _) = Libp2pService::start(
+        addr.clone(),
+        &[Libp2pStrategy::MulticastDns],
+        &[],
+        std::time::Duration::from_secs(10),
+    )
+    .await
+    .unwrap();
 
     let addr2: Multiaddr = test_utils::make_address("/ip6/::1/tcp/");
-    let (mut serv2, _) = Libp2pService::start(addr2.clone(), &[Libp2pStrategy::MulticastDns], &[])
-        .await
-        .unwrap();
+    let (mut serv2, _) = Libp2pService::start(
+        addr2.clone(),
+        &[Libp2pStrategy::MulticastDns],
+        &[],
+        std::time::Duration::from_secs(10),
+    )
+    .await
+    .unwrap();
 
     loop {
         let (serv_res, _) = tokio::join!(serv.poll_next(), serv2.poll_next());
@@ -73,12 +84,24 @@ async fn test_libp2p_peer_discovery() {
 #[tokio::test]
 async fn test_libp2p_floodsub() {
     let addr1: Multiaddr = test_utils::make_address("/ip6/::1/tcp/");
-    let (mut conn1, mut flood1) =
-        Libp2pService::start(addr1, &[], &[FloodsubTopic::Transactions]).await.unwrap();
+    let (mut conn1, mut flood1) = Libp2pService::start(
+        addr1,
+        &[],
+        &[FloodsubTopic::Transactions],
+        std::time::Duration::from_secs(10),
+    )
+    .await
+    .unwrap();
 
     let addr2: Multiaddr = test_utils::make_address("/ip6/::1/tcp/");
-    let (mut conn2, mut flood2) =
-        Libp2pService::start(addr2, &[], &[FloodsubTopic::Transactions]).await.unwrap();
+    let (mut conn2, mut flood2) = Libp2pService::start(
+        addr2,
+        &[],
+        &[FloodsubTopic::Transactions],
+        std::time::Duration::from_secs(10),
+    )
+    .await
+    .unwrap();
 
     let (conn1_res, conn2_res) =
         tokio::join!(conn1.connect(conn2.local_addr().clone()), conn2.poll_next());
diff --git a/p2p/tests/p2p.rs b/p2p/tests/p2p.rs
index 0fd081ed5..3a5eb5d8d 100644
--- a/p2p/tests/p2p.rs
+++ b/p2p/tests/p2p.rs
@@ -14,6 +14,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+#![cfg(not(loom))]
 extern crate test_utils;
 
 use common::{chain::config, sync::Arc};
diff --git a/p2p/tests/peer.rs b/p2p/tests/peer.rs
index 86d6afdae..891ad9fef 100644
--- a/p2p/tests/peer.rs
+++ b/p2p/tests/peer.rs
@@ -14,6 +14,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+#![cfg(not(loom))]
 extern crate test_utils;
 
 use common::chain::config;
@@ -31,7 +32,9 @@ use tokio::net::TcpStream;
 async fn test_peer_new_mock() {
     let config = Arc::new(config::create_mainnet());
     let addr: SocketAddr = test_utils::make_address("[::1]:");
-    let (mut server, _) = MockService::start(addr, &[], &[]).await.unwrap();
+    let (mut server, _) = MockService::start(addr, &[], &[], std::time::Duration::from_secs(10))
+        .await
+        .unwrap();
     let peer_fut = TcpStream::connect(addr);
 
     let (server_res, peer_res) = tokio::join!(server.poll_next(), peer_fut);
@@ -63,11 +66,17 @@ async fn test_peer_new_mock() {
 async fn test_peer_new_libp2p() {
     let config = Arc::new(config::create_mainnet());
     let addr1: Multiaddr = test_utils::make_address("/ip6/::1/tcp/");
-    let (mut server1, _) = Libp2pService::start(addr1.clone(), &[], &[]).await.unwrap();
+    let (mut server1, _) =
+        Libp2pService::start(addr1.clone(), &[], &[], std::time::Duration::from_secs(10))
+            .await
+            .unwrap();
 
     let conn_addr = server1.local_addr().clone();
     let addr2: Multiaddr = test_utils::make_address("/ip6/::1/tcp/");
-    let (mut server2, _) = Libp2pService::start(addr2, &[], &[]).await.unwrap();
+    let (mut server2, _) =
+        Libp2pService::start(addr2, &[], &[], std::time::Duration::from_secs(10))
+            .await
+            .unwrap();
 
     let (server1_res, server2_res) = tokio::join!(server1.poll_next(), server2.connect(conn_addr));
     assert!(server1_res.is_ok());