From 8ab290ea03ea4fd61fce0280f3ef9aa87531a8b6 Mon Sep 17 00:00:00 2001 From: mfahampshire Date: Wed, 29 Apr 2026 12:36:50 +0100 Subject: [PATCH] Remove old libp2p code - new example usage coming in smolmix-companions-docs branch --- sdk/rust/nym-sdk/Cargo.toml | 2 - .../nym-sdk/examples/libp2p_chat/README.md | 64 -- sdk/rust/nym-sdk/examples/libp2p_chat/main.rs | 173 ---- .../nym-sdk/examples/libp2p_ping/README.md | 79 -- sdk/rust/nym-sdk/examples/libp2p_ping/main.rs | 143 --- .../examples/libp2p_shared/connection.rs | 418 -------- .../nym-sdk/examples/libp2p_shared/error.rs | 64 -- .../nym-sdk/examples/libp2p_shared/lib.rs | 10 - .../nym-sdk/examples/libp2p_shared/message.rs | 335 ------- .../nym-sdk/examples/libp2p_shared/mixnet.rs | 164 ---- .../nym-sdk/examples/libp2p_shared/queue.rs | 138 --- .../examples/libp2p_shared/substream.rs | 412 -------- .../examples/libp2p_shared/transport.rs | 898 ------------------ 13 files changed, 2900 deletions(-) delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_chat/README.md delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_chat/main.rs delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_ping/README.md delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_ping/main.rs delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_shared/connection.rs delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_shared/error.rs delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_shared/lib.rs delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_shared/message.rs delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_shared/mixnet.rs delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_shared/queue.rs delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_shared/substream.rs delete mode 100644 sdk/rust/nym-sdk/examples/libp2p_shared/transport.rs diff --git a/sdk/rust/nym-sdk/Cargo.toml b/sdk/rust/nym-sdk/Cargo.toml index 931c640edb..1b6760d975 100644 --- a/sdk/rust/nym-sdk/Cargo.toml +++ b/sdk/rust/nym-sdk/Cargo.toml @@ -97,8 +97,6 @@ tokio = { workspace = true, features = ["full", "test-util"] } time = { workspace = true } nym-bin-common = {workspace = true, features = ["basic_tracing"] } -# extra dependencies for libp2p examples -#libp2p = { git = "https://github.com/ChainSafe/rust-libp2p.git", rev = "e3440d25681df380c9f0f8cfdcfd5ecc0a4f2fb6", features = [ "identify", "macros", "ping", "tokio", "tcp", "dns", "websocket", "noise", "mplex", "yamux", "gossipsub" ]} tokio-stream = { workspace = true } tokio-util = { workspace = true, features = ["codec"] } parking_lot = { workspace = true } diff --git a/sdk/rust/nym-sdk/examples/libp2p_chat/README.md b/sdk/rust/nym-sdk/examples/libp2p_chat/README.md deleted file mode 100644 index 0287329769..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_chat/README.md +++ /dev/null @@ -1,64 +0,0 @@ -# rust-libp2p-nym - -This repo contains an example implementation of a libp2p transport using the Nym mixnet. It relies on the ChainSafe's fork of libp2p: https://github.com/ChainSafe/rust-libp2p - -## Requirements - -- Rust 1.68.2 -- `Protoc` protobuf compiler. On Debian/Ubuntu distributed via `apt` as `protobuf-compiler` & on Arch/Manjaro via AUR as `[python-protobuf-compiler](https://aur.archlinux.org/packages/python-protobuf-compiler)`. - -## Usage - -To instantiate a libp2p swarm using the transport: - -```rust -use libp2p::core::{muxing::StreamMuxerBox, transport::Transport}; -use libp2p::swarm::{keep_alive::Behaviour, SwarmBuilder}; -use libp2p::{identity, PeerId}; -use nym_sdk::mixnet::MixnetClient; -use rust_libp2p_nym::transport::NymTransport; -use rust_libp2p_nym::test_utils::create_nym_client; -use std::error::Error; - -#[tokio::main] -async fn main() -> Result<(), Box> { - let local_key = identity::Keypair::generate_ed25519(); - let local_peer_id = PeerId::from(local_key.public()); - info!("Local peer id: {local_peer_id:?}"); - - let nym_id = rand::random::().to_string(); - let nym_client = MixnetClient::connect_new().await.unwrap(); - let transport = NymTransport::new(nym_client, local_key.clone()).await?; - let _swarm = SwarmBuilder::with_tokio_executor( - transport - .map(|a, _| (a.0, StreamMuxerBox::new(a.1))) - .boxed(), - Behaviour::default(), - local_peer_id, - ) - .build(); - Ok(()) -} -``` - -## Chat example - -To run the libp2p chat example, run the following in one terminal: -```bash -cargo run --example libp2p_chat -# Local peer id: PeerId("12D3KooWLukBu6q2FerWPFhFFhiYaJkhn2sBmceh9UCaXe6hJf5D") -# Listening on "/nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve" -``` - -In another terminal, run ping again, passing the Nym multiaddress printed previously: -```bash -cargo run --example libp2p_chat -- /nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve -# Local peer id: PeerId("12D3KooWNsuRwG6DHnFJCDR8B3zdvja6xLcfnbtKCsQWJ8eppyWC") -# Dialed /nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve -# Listening on "/nym/2oiRW5C9ivyF3Bo3Gpm4H9EqSKH7A6GpcrRRwVSDVUQ9.EajgCnhzimsP6KskUwKcEj8VFCmHR78s2J6FHWcZ4etR@Fo4f4SQLdoyoGkFae5TpVhRVoXCF8UiypLVGtGjujVPf" -``` - -You should see that the nodes connected and sent messages to each other: -```bash -# 2023-08-10T14:06:28.116Z INFO libp2p_chat > Got message: 'hello world' with id: 37393732353836333838333537303637303237 from peer: 12D3KooWB6k8ZGDF44N4FMRhgVBNihwk1wMYSumosxiZq9pUTbAz -``` diff --git a/sdk/rust/nym-sdk/examples/libp2p_chat/main.rs b/sdk/rust/nym-sdk/examples/libp2p_chat/main.rs deleted file mode 100644 index 66cd9ea0e5..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_chat/main.rs +++ /dev/null @@ -1,173 +0,0 @@ -// Copyright 2018 Parity Technologies (UK) Ltd. -// -// Permission is hereby granted, free of charge, to any person obtaining a -// copy of this software and associated documentation files (the "Software"), -// to deal in the Software without restriction, including without limitation -// the rights to use, copy, modify, merge, publish, distribute, sublicense, -// and/or sell copies of the Software, and to permit persons to whom the -// Software is furnished to do so, subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -// DEALINGS IN THE SOFTWARE. - -//! A basic chat application with logs demonstrating libp2p and the gossipsub protocol -//! combined with mDNS for the discovery of peers to gossip with. -//! -//! Using two terminal windows, start two instances, typing the following in each: -//! -//! ```sh -//! cargo run -//! ``` -//! -//! Mutual mDNS discovery may take a few seconds. When each peer does discover the other -//! it will print a message like: -//! -//! ```sh -//! mDNS discovered a new peer: {peerId} -//! ``` -//! -//! Type a message and hit return: the message is sent and printed in the other terminal. -//! Close with Ctrl-c. -//! -//! You can open more terminal windows and add more peers using the same line above. -//! -//! Once an additional peer is mDNS discovered it can participate in the conversation -//! and all peers will receive messages sent from it. -//! -//! If a participant exits (Control-C or otherwise) the other peers will receive an mDNS expired -//! event and remove the expired peer from the list of known peers. - -// use crate::rust_libp2p_nym::transport::NymTransport; -// use futures::{prelude::*, select}; -// use libp2p::Multiaddr; -// use libp2p::{ -// core::muxing::StreamMuxerBox, -// gossipsub, identity, -// swarm::NetworkBehaviour, -// swarm::{SwarmBuilder, SwarmEvent}, -// PeerId, Transport, -// }; -// use log::{error, info, LevelFilter}; -// use nym_sdk::mixnet::MixnetClient; -// use std::collections::hash_map::DefaultHasher; -use std::error::Error; -// use std::hash::{Hash, Hasher}; -// use std::time::Duration; -// use tokio::io; -// use tokio_util::codec; - -// #[path = "../libp2p_shared/lib.rs"] -// mod rust_libp2p_nym; -// -// // We create a custom network behaviour that uses Gossipsub -// #[derive(NetworkBehaviour)] -// struct Behaviour { -// gossipsub: gossipsub::Behaviour, -// } - -#[tokio::main] -async fn main() -> Result<(), Box> { - unimplemented!("temporarily disabled") - - // pretty_env_logger::formatted_timed_builder() - // .filter_level(LevelFilter::Warn) - // .filter(Some("libp2p_chat"), LevelFilter::Info) - // .init(); - // - // // Create a random PeerId - // let id_keys = identity::Keypair::generate_ed25519(); - // let local_peer_id = PeerId::from(id_keys.public()); - // info!("Local peer id: {local_peer_id}"); - // - // // To content-address message, we can take the hash of message and use it as an ID. - // let message_id_fn = |message: &gossipsub::Message| { - // let mut s = DefaultHasher::new(); - // message.data.hash(&mut s); - // gossipsub::MessageId::from(s.finish().to_string()) - // }; - // - // // Set a custom gossipsub configuration - // let gossipsub_config = gossipsub::ConfigBuilder::default() - // .heartbeat_interval(Duration::from_secs(10)) // This is set to aid debugging by not cluttering the log space - // .validation_mode(gossipsub::ValidationMode::Strict) // This sets the kind of message validation. The default is Strict (enforce message signing) - // .message_id_fn(message_id_fn) // content-address messages. No two messages of the same content will be propagated. - // .build() - // .expect("Valid config"); - // - // // build a gossipsub network behaviour - // let mut gossipsub = gossipsub::Behaviour::new( - // gossipsub::MessageAuthenticity::Signed(id_keys), - // gossipsub_config, - // ) - // .expect("Correct configuration"); - // // Create a Gossipsub topic - // let topic = gossipsub::IdentTopic::new("test-net"); - // // subscribes to our topic - // gossipsub.subscribe(&topic)?; - // - // let client = MixnetClient::connect_new().await.unwrap(); - // info!("client address: {}", client.nym_address()); - // - // let local_key = identity::Keypair::generate_ed25519(); - // let local_peer_id = PeerId::from(local_key.public()); - // info!("Local peer id: {local_peer_id:?}"); - // - // let transport = NymTransport::new(client, local_key).await?; - // - // let mut swarm = SwarmBuilder::with_tokio_executor( - // transport - // .map(|a, _| (a.0, StreamMuxerBox::new(a.1))) - // .boxed(), - // Behaviour { gossipsub }, - // local_peer_id, - // ) - // .build(); - // - // if let Some(addr) = std::env::args().nth(1) { - // let remote: Multiaddr = addr.parse()?; - // swarm.dial(remote)?; - // info!("Dialed {addr}") - // } - // - // // Read full lines from stdin - // let mut stdin = codec::FramedRead::new(io::stdin(), codec::LinesCodec::new()).fuse(); - // - // info!("Enter messages via STDIN and they will be sent to connected peers using Gossipsub"); - // - // // Kick it off - // loop { - // select! { - // line = stdin.select_next_some() => { - // if let Err(e) = swarm - // .behaviour_mut().gossipsub - // .publish(topic.clone(), line.expect("Stdin not to close").as_bytes()) { - // error!("Publish error: {e:?}"); - // } - // }, - // event = swarm.select_next_some() => { - // match event { - // SwarmEvent::Behaviour(BehaviourEvent::Gossipsub(gossipsub::Event::Message { - // propagation_source: peer_id, - // message_id: id, - // message, - // })) => info!( - // "Got message: '{}' with id: {id} from peer: {peer_id}", - // String::from_utf8_lossy(&message.data), - // ), - // SwarmEvent::NewListenAddr { address, .. } => { - // info!("Local node is listening on {address}"); - // } - // other => {info!("other event: {:?}", other)} - // } - // } - // } - // } -} diff --git a/sdk/rust/nym-sdk/examples/libp2p_ping/README.md b/sdk/rust/nym-sdk/examples/libp2p_ping/README.md deleted file mode 100644 index 2218793e89..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_ping/README.md +++ /dev/null @@ -1,79 +0,0 @@ -# rust-libp2p-nym - -This repo contains an example implementation of a libp2p transport using the Nym mixnet. It relies on the ChainSafe's fork of libp2p: https://github.com/ChainSafe/rust-libp2p - -## Requirements - -- Rust 1.68.2 -- `Protoc` protobuf compiler. On Debian/Ubuntu distributed via `apt` as `protobuf-compiler` & on Arch/Manjaro via AUR as `[python-protobuf-compiler](https://aur.archlinux.org/packages/python-protobuf-compiler)`. - -## Usage - -To instantiate a libp2p swarm using the transport: - -```rust -use libp2p::core::{muxing::StreamMuxerBox, transport::Transport}; -use libp2p::swarm::{keep_alive::Behaviour, SwarmBuilder}; -use libp2p::{identity, PeerId}; -use nym_sdk::mixnet::MixnetClient; -use rust_libp2p_nym::transport::NymTransport; -use rust_libp2p_nym::test_utils::create_nym_client; -use std::error::Error; - -#[tokio::main] -async fn main() -> Result<(), Box> { - let local_key = identity::Keypair::generate_ed25519(); - let local_peer_id = PeerId::from(local_key.public()); - info!("Local peer id: {local_peer_id:?}"); - - let nym_id = rand::random::().to_string(); - let nym_client = MixnetClient::connect_new().await.unwrap(); - let transport = NymTransport::new(nym_client, local_key.clone()).await?; - let _swarm = SwarmBuilder::with_tokio_executor( - transport - .map(|a, _| (a.0, StreamMuxerBox::new(a.1))) - .boxed(), - Behaviour::default(), - local_peer_id, - ) - .build(); - Ok(()) -} -``` - -## Ping example - -To run the libp2p ping example, run the following in one terminal: -```bash -cargo run --example libp2p_ping -# Local peer id: PeerId("12D3KooWLukBu6q2FerWPFhFFhiYaJkhn2sBmceh9UCaXe6hJf5D") -# Listening on "/nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve" -``` - -In another terminal, run ping again, passing the Nym multiaddress printed previously: -```bash -cargo run --example libp2p_ping -- /nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve -# Local peer id: PeerId("12D3KooWNsuRwG6DHnFJCDR8B3zdvja6xLcfnbtKCsQWJ8eppyWC") -# Dialed /nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve -# Listening on "/nym/2oiRW5C9ivyF3Bo3Gpm4H9EqSKH7A6GpcrRRwVSDVUQ9.EajgCnhzimsP6KskUwKcEj8VFCmHR78s2J6FHWcZ4etR@Fo4f4SQLdoyoGkFae5TpVhRVoXCF8UiypLVGtGjujVPf" -``` - -You should see that the nodes connected and pinged each other: -```bash -# Mar 30 22:56:36.400 INFO ping: BehaviourEvent: Event { peer: PeerId("12D3KooWGf2oYd6U2nrLzfDrN9zxsjSQjPsMh2oDJPUQ9hiHMNtf"), result: Ok(Ping { rtt: 1.06836675s }) } -``` -```bash -# Mar 30 22:56:35.595 INFO ping: BehaviourEvent: Event { peer: PeerId("12D3KooWMd5ak31DXuZq7x1JuFSR6toA5RDQrPaHrfXEhy7vqqpC"), result: Ok(Pong) } -``` - -In order to run the ping example with vanilla libp2p, which uses tcp, pass the -`--features libp2p-vanilla` flag to the example and follow the instructions on the -rust-libp2p project as usual. - -```bash -RUST_LOG=ping=debug cargo run --example ping --features libp2p-vanilla -``` - -```bash -RUST_LOG=ping=debug cargo run --example ping --features libp2p-vanilla -- "/ip4/127.0.0.1/tcp/$PORT" -``` diff --git a/sdk/rust/nym-sdk/examples/libp2p_ping/main.rs b/sdk/rust/nym-sdk/examples/libp2p_ping/main.rs deleted file mode 100644 index 06c8569803..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_ping/main.rs +++ /dev/null @@ -1,143 +0,0 @@ -// Copyright 2018 Parity Technologies (UK) Ltd. -// -// Permission is hereby granted, free of charge, to any person obtaining a -// copy of this software and associated documentation files (the "Software"), -// to deal in the Software without restriction, including without limitation -// the rights to use, copy, modify, merge, publish, distribute, sublicense, -// and/or sell copies of the Software, and to permit persons to whom the -// Software is furnished to do so, subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -// DEALINGS IN THE SOFTWARE. - -//! Ping example -//! -//! See ../src/tutorial.rs for a step-by-step guide building the example below. -//! -//! In the first terminal window, run: -//! -//! ```sh -//! cargo run --example ping --features=full -//! ``` -//! -//! It will print the PeerId and the listening addresses, e.g. `Listening on -//! "/ip4/0.0.0.0/tcp/24915"` -//! -//! In the second terminal window, start a new instance of the example with: -//! -//! ```sh -//! cargo run --example ping --features=full -- /ip4/127.0.0.1/tcp/24915 -//! ``` -//! -//! The two nodes establish a connection, negotiate the ping protocol -//! and begin pinging each other. - -// use libp2p::futures::StreamExt; -// use libp2p::ping::Success; -// use libp2p::swarm::{keep_alive, NetworkBehaviour, SwarmEvent}; -// use libp2p::{identity, ping, Multiaddr, PeerId}; -// use log::{debug, info, LevelFilter}; -use std::error::Error; -// use std::time::Duration; -// -// #[path = "../libp2p_shared/lib.rs"] -// mod rust_libp2p_nym; - -#[tokio::main] -async fn main() -> Result<(), Box> { - unimplemented!("temporarily disabled") - // - // pretty_env_logger::formatted_timed_builder() - // .filter_level(LevelFilter::Warn) - // .filter(Some("libp2p_ping"), LevelFilter::Debug) - // .init(); - // - // let local_key = identity::Keypair::generate_ed25519(); - // let local_peer_id = PeerId::from(local_key.public()); - // info!("Local peer id: {local_peer_id:?}"); - // - // #[cfg(not(feature = "libp2p-vanilla"))] - // let mut swarm = { - // debug!("Running `ping` example using NymTransport"); - // use libp2p::core::{muxing::StreamMuxerBox, transport::Transport}; - // use libp2p::swarm::SwarmBuilder; - // use rust_libp2p_nym::transport::NymTransport; - // - // let client = nym_sdk::mixnet::MixnetClient::connect_new().await.unwrap(); - // - // let transport = NymTransport::new(client, local_key.clone()).await?; - // SwarmBuilder::with_tokio_executor( - // transport - // .map(|a, _| (a.0, StreamMuxerBox::new(a.1))) - // .boxed(), - // Behaviour::default(), - // local_peer_id, - // ) - // .build() - // }; - // - // #[cfg(feature = "libp2p-vanilla")] - // let mut swarm = { - // debug!("Running `ping` example using the vanilla libp2p tokio_development_transport"); - // let transport = libp2p::tokio_development_transport(local_key)?; - // let mut swarm = - // libp2p::Swarm::with_tokio_executor(transport, Behaviour::default(), local_peer_id); - // swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?; - // swarm - // }; - // - // // Dial the peer identified by the multi-address given as the second - // // command-line argument, if any. - // if let Some(addr) = std::env::args().nth(1) { - // let remote: Multiaddr = addr.parse()?; - // swarm.dial(remote)?; - // info!("Dialed {addr}") - // } - // - // let mut total_ping_rtt: Duration = Duration::from_micros(0); - // let mut counter: u128 = 0; - // loop { - // match swarm.select_next_some().await { - // SwarmEvent::NewListenAddr { address, .. } => info!("Listening on {address:?}"), - // SwarmEvent::Behaviour(event) => { - // // Get the round-trip duration for the pings. - // // This value is already captured in the BehaviourEvent::Ping's `Success::Ping` - // // field. - // debug!("{event:?}"); - // if let BehaviourEvent::Ping(ping_event) = event { - // let result: Success = ping_event.result?; - // match result { - // Success::Ping { rtt } => { - // counter += 1; - // total_ping_rtt += rtt; - // let average_ping_rtt = Duration::from_micros( - // (total_ping_rtt.as_micros() / counter).try_into().unwrap(), - // ); - // info!("Ping RTT: {rtt:?} AVERAGE RTT: ({counter} pings): {average_ping_rtt:?}"); - // } - // Success::Pong => info!("Pong Event"), - // } - // } - // } - // _ => {} - // } - // } -} -// -// /// Our network behaviour. -// /// -// /// For illustrative purposes, this includes the [`KeepAlive`](behaviour::KeepAlive) behaviour so a continuous sequence of -// /// pings can be observed. -// #[derive(NetworkBehaviour, Default)] -// struct Behaviour { -// keep_alive: keep_alive::Behaviour, -// ping: ping::Behaviour, -// } diff --git a/sdk/rust/nym-sdk/examples/libp2p_shared/connection.rs b/sdk/rust/nym-sdk/examples/libp2p_shared/connection.rs deleted file mode 100644 index d562d634de..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_shared/connection.rs +++ /dev/null @@ -1,418 +0,0 @@ -use libp2p::core::{muxing::StreamMuxerEvent, PeerId, StreamMuxer}; -use log::debug; -use nym_sphinx::addressing::clients::Recipient; -use std::{ - collections::{HashMap, HashSet}, - pin::Pin, - sync::{ - atomic::{AtomicU64, Ordering}, - Arc, - }, - task::{Context, Poll, Waker}, -}; -use tokio::sync::{ - mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}, - oneshot, -}; - -use super::error::Error; -use super::message::{ - ConnectionId, Message, OutboundMessage, SubstreamId, SubstreamMessage, SubstreamMessageType, - TransportMessage, -}; -use super::substream::Substream; - -/// Connection represents the result of a connection setup process. -/// It implements `StreamMuxer` and thus has stream multiplexing built in. -#[derive(Debug)] -pub struct Connection { - pub(crate) peer_id: PeerId, - pub(crate) remote_recipient: Recipient, - pub(crate) id: ConnectionId, - - /// receive inbound messages from the `InnerConnection` - pub(crate) inbound_rx: UnboundedReceiver, - - /// substream ID -> outbound pending substream exists - /// the key is deleted when the response is received, or the request times out - pending_substreams: HashSet, - - /// substream ID -> substream's inbound_tx channel - substream_inbound_txs: HashMap>>, - - /// substream ID -> substream's close_tx channel - substream_close_txs: HashMap>, - - /// send messages to the mixnet - /// used for sending `SubstreamMessageType::OpenRequest` messages - /// also passed to each substream so they can write to the mixnet - pub(crate) mixnet_outbound_tx: UnboundedSender, - - /// inbound substream open requests; used in poll_inbound - inbound_open_tx: UnboundedSender, - inbound_open_rx: UnboundedReceiver, - - /// closed substream IDs; used in poll_close - close_tx: UnboundedSender, - close_rx: UnboundedReceiver, - - /// message nonce contains the next nonce that should be used when - /// sending a message over the connection - pub(crate) message_nonce: Arc, - - waker: Option, -} - -impl Connection { - pub(crate) fn new( - peer_id: PeerId, - remote_recipient: Recipient, - id: ConnectionId, - inbound_rx: UnboundedReceiver, - mixnet_outbound_tx: UnboundedSender, - ) -> Self { - let (inbound_open_tx, inbound_open_rx) = unbounded_channel(); - let (close_tx, close_rx) = unbounded_channel(); - - Connection { - peer_id, - remote_recipient, - id, - inbound_rx, - pending_substreams: HashSet::new(), - substream_inbound_txs: HashMap::new(), - substream_close_txs: HashMap::new(), - mixnet_outbound_tx, - inbound_open_tx, - inbound_open_rx, - close_tx, - close_rx, - message_nonce: Arc::new(AtomicU64::new(1)), - waker: None, - } - } - - fn new_outbound_substream(&mut self) -> Result { - let substream_id = SubstreamId::generate(); - let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst); - - // send the substream open request that requests to open a substream with the given ID - self.mixnet_outbound_tx - .send(OutboundMessage { - recipient: self.remote_recipient, - message: Message::TransportMessage(TransportMessage { - nonce, - id: self.id.clone(), - message: SubstreamMessage { - substream_id: substream_id.clone(), - message_type: SubstreamMessageType::OpenRequest, - }, - }), - }) - .map_err(|e| Error::OutboundSendFailure(e.to_string()))?; - - // track pending outbound substreams - // TODO we should probably lock this? storing map values should be atomic - let res = self.new_substream(substream_id.clone()); - if res.is_ok() { - self.pending_substreams.insert(substream_id); - } - res - } - - // creates a new substream instance with the given ID. - fn new_substream(&mut self, id: SubstreamId) -> Result { - // check we don't already have a substream with this ID - if self.substream_inbound_txs.contains_key(&id) { - return Err(Error::SubstreamIdExists(id)); - } - - let (inbound_tx, inbound_rx) = unbounded_channel::>(); - let (close_tx, close_rx) = oneshot::channel::<()>(); - self.substream_inbound_txs.insert(id.clone(), inbound_tx); - self.substream_close_txs.insert(id.clone(), close_tx); - - if let Some(waker) = self.waker.take() { - waker.wake(); - } - - Ok(Substream::new( - self.remote_recipient, - self.id.clone(), - id, - inbound_rx, - self.mixnet_outbound_tx.clone(), - close_rx, - self.message_nonce.clone(), - )) - } - - fn handle_close(&mut self, substream_id: SubstreamId) -> Result<(), Error> { - if self.substream_inbound_txs.remove(&substream_id).is_none() { - return Err(Error::SubstreamIdDoesNotExist(substream_id)); - } - - // notify substream that it's closed - let close_tx = self.substream_close_txs.remove(&substream_id); - close_tx.unwrap().send(()).unwrap(); - - // notify poll_close that the substream is closed - self.close_tx - .send(substream_id) - .map_err(|e| Error::InboundSendFailure(e.to_string())) - } -} - -impl StreamMuxer for Connection { - type Substream = Substream; - type Error = Error; - - fn poll_inbound( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll> { - if let Poll::Ready(Some(substream)) = self.inbound_open_rx.poll_recv(cx) { - return Poll::Ready(Ok(substream)); - } - - Poll::Pending - } - - fn poll_outbound( - mut self: Pin<&mut Self>, - _cx: &mut Context<'_>, - ) -> Poll> { - Poll::Ready(self.new_outbound_substream()) - } - - fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - if let Poll::Ready(Some(_)) = self.close_rx.poll_recv(cx) { - return Poll::Ready(Ok(())); - } - - Poll::Pending - } - - fn poll( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll> { - while let Poll::Ready(Some(msg)) = self.inbound_rx.poll_recv(cx) { - match msg.message_type { - SubstreamMessageType::OpenRequest => { - // create a new substream with the given ID - let substream = self.new_substream(msg.substream_id.clone())?; - let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst); - - // send the response to the remote peer - self.mixnet_outbound_tx - .send(OutboundMessage { - recipient: self.remote_recipient, - message: Message::TransportMessage(TransportMessage { - nonce, - id: self.id.clone(), - message: SubstreamMessage { - substream_id: msg.substream_id.clone(), - message_type: SubstreamMessageType::OpenResponse, - }, - }), - }) - .map_err(|e| Error::OutboundSendFailure(e.to_string()))?; - debug!("wrote OpenResponse for substream: {:?}", &msg.substream_id); - - // send the substream to our own channel to be returned in poll_inbound - self.inbound_open_tx - .send(substream) - .map_err(|e| Error::InboundSendFailure(e.to_string()))?; - - debug!("new inbound substream: {:?}", &msg.substream_id); - } - SubstreamMessageType::OpenResponse => { - if !self.pending_substreams.remove(&msg.substream_id) { - debug!( - "SubstreamMessageType::OpenResponse no substream pending for ID: {:?}", - &msg.substream_id - ); - } - } - SubstreamMessageType::Close => { - self.handle_close(msg.substream_id)?; - } - SubstreamMessageType::Data(data) => { - debug!("SubstreamMessageType::Data: {:?}", &data); - let inbound_tx = self - .substream_inbound_txs - .get_mut(&msg.substream_id) - .expect("must have a substream channel for substream"); - - // NOTE: this ignores channel closed errors, which is fine because the substream - // might have been closed/dropped - inbound_tx.send(data).ok(); - } - } - } - - self.waker = Some(cx.waker().clone()); - Poll::Pending - } -} - -/// PendingConnection represents a connection that's been initiated, but not completed. -pub(crate) struct PendingConnection { - pub(crate) remote_recipient: Recipient, - pub(crate) connection_tx: oneshot::Sender, -} - -impl PendingConnection { - pub(crate) fn new( - remote_recipient: Recipient, - connection_tx: oneshot::Sender, - ) -> Self { - PendingConnection { - remote_recipient, - connection_tx, - } - } -} - -#[cfg(test)] -mod test { - use super::super::message::InboundMessage; - use super::super::mixnet::initialize_mixnet; - use super::*; - use futures::future::poll_fn; - use futures::{AsyncReadExt, AsyncWriteExt, FutureExt}; - use nym_sdk::mixnet::MixnetClient; - - async fn inbound_receive_and_send( - connection_id: ConnectionId, - mixnet_inbound_rx: &mut UnboundedReceiver, - inbound_tx: &UnboundedSender, - expected_nonce: u64, - ) { - let recv_msg = mixnet_inbound_rx.recv().await.unwrap(); - match recv_msg.0 { - Message::TransportMessage(TransportMessage { - nonce, - id, - message: msg, - }) => { - assert_eq!(nonce, expected_nonce); - assert_eq!(id, connection_id); - inbound_tx.send(msg).unwrap(); - } - _ => panic!("unexpected message"), - } - } - - #[tokio::test] - async fn test_connection_stream_muxer() { - let client = MixnetClient::connect_new().await.unwrap(); - let (sender_address, mut sender_mixnet_inbound_rx, sender_outbound_tx) = - initialize_mixnet(client, None).await.unwrap(); - - let client2 = MixnetClient::connect_new().await.unwrap(); - - let (recipient_address, mut recipient_mixnet_inbound_rx, recipient_outbound_tx) = - initialize_mixnet(client2, None).await.unwrap(); - - let connection_id = ConnectionId::generate(); - - let recipient_peer_id = PeerId::random(); - let sender_peer_id = PeerId::random(); - - // create the connections - let (sender_inbound_tx, sender_inbound_rx) = unbounded_channel::(); - let mut sender_connection = Connection::new( - recipient_peer_id, - recipient_address, - connection_id.clone(), - sender_inbound_rx, - sender_outbound_tx, - ); - let (recipient_inbound_tx, recipient_inbound_rx) = unbounded_channel::(); - let mut recipient_connection = Connection::new( - sender_peer_id, - sender_address, - connection_id.clone(), - recipient_inbound_rx, - recipient_outbound_tx, - ); - - // send the substream OpenRequest to the mixnet - let mut sender_substream = sender_connection.new_outbound_substream().unwrap(); - assert!(sender_connection - .pending_substreams - .contains(&sender_substream.substream_id)); - assert_eq!(sender_connection.message_nonce.load(Ordering::SeqCst), 2); - - // poll the recipient inbound stream; should receive the OpenRequest and create the substream - inbound_receive_and_send( - connection_id.clone(), - &mut recipient_mixnet_inbound_rx, - &recipient_inbound_tx, - 1, - ) - .await; - poll_fn(|cx| Pin::new(&mut recipient_connection).as_mut().poll(cx)).now_or_never(); - assert_eq!(recipient_connection.message_nonce.load(Ordering::SeqCst), 2); - - // poll recipient's poll_inbound to receive the substream - let maybe_recipient_substream = poll_fn(|cx| { - Pin::new(&mut recipient_connection) - .as_mut() - .poll_inbound(cx) - }) - .now_or_never(); - let mut recipient_substream = maybe_recipient_substream.unwrap().unwrap(); - - // poll sender's connection to receive the OpenResponse and send it to the Connection inbound channel - inbound_receive_and_send( - connection_id.clone(), - &mut sender_mixnet_inbound_rx, - &sender_inbound_tx, - 1, - ) - .await; - - // poll sender's poll_outbound to get the substream - poll_fn(|cx| Pin::new(&mut sender_connection).as_mut().poll(cx)).now_or_never(); - assert!(sender_connection.pending_substreams.is_empty()); - - // finally, write message to the substream - let data = b"hello world"; - sender_substream.write_all(data).await.unwrap(); - assert_eq!(sender_connection.message_nonce.load(Ordering::SeqCst), 3); - - // receive message from the mixnet, push to the recipient Connection inbound channel - inbound_receive_and_send( - connection_id.clone(), - &mut recipient_mixnet_inbound_rx, - &recipient_inbound_tx, - 2, - ) - .await; - - // poll the sender's connection to send the msg from the connection inbound channel to the substream's - poll_fn(|cx| Pin::new(&mut sender_connection).as_mut().poll(cx)).now_or_never(); - - // poll the recipient's connection to read the msg from the mixnet and mux it into the substream - poll_fn(|cx| Pin::new(&mut recipient_connection).as_mut().poll(cx)).now_or_never(); - - let mut buf = [0u8; 11]; - let n = recipient_substream.read(&mut buf).await.unwrap(); - assert_eq!(n, 11); - assert_eq!(buf, data[..]); - - // test closing the stream; assert the stream is closed on both sides - sender_substream.close().await.unwrap(); - assert_eq!(sender_connection.message_nonce.load(Ordering::SeqCst), 4); - inbound_receive_and_send( - connection_id.clone(), - &mut recipient_mixnet_inbound_rx, - &recipient_inbound_tx, - 3, - ) - .await; - } -} diff --git a/sdk/rust/nym-sdk/examples/libp2p_shared/error.rs b/sdk/rust/nym-sdk/examples/libp2p_shared/error.rs deleted file mode 100644 index a81674b0fa..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_shared/error.rs +++ /dev/null @@ -1,64 +0,0 @@ -use libp2p::core::multiaddr; -use nym_sphinx::addressing::clients::RecipientFormattingError; - -use super::message::SubstreamId; - -#[derive(Debug, thiserror::Error)] -pub enum Error { - #[error("unimplemented")] - Unimplemented, - #[error("failed to format multiaddress from nym address")] - FailedToFormatMultiaddr(#[from] multiaddr::Error), - #[error("unexpected protocol in multiaddress")] - InvalidProtocolForMultiaddr, - #[error("failed to decode message")] - InvalidMessageBytes, - #[error("no connection found for ConnectionResponse")] - NoConnectionForResponse, - #[error("received ConnectionResponse but connection was already established")] - ConnectionAlreadyEstablished, - #[error("received None recipient in ConnectionRequest")] - NoneRecipientInConnectionRequest, - #[error("cannot handle connection request; already have connection with given ID")] - ConnectionIDExists, - #[error("no connection found for TransportMessage")] - NoConnectionForTransportMessage, - #[error("failed to decode ConnectionMessage; too short")] - ConnectionMessageBytesTooShort, - #[error("failed to decode ConnectionMessage; no recipient")] - ConnectionMessageBytesNoRecipient, - #[error("failed to decode ConnectionMessage; no peer ID")] - ConnectionMessageBytesNoPeerId, - #[error("invalid peer ID bytes")] - InvalidPeerIdBytes, - #[error("invalid recipient bytes")] - InvalidRecipientBytes(#[from] RecipientFormattingError), - #[error("invalid recipient prefix byte")] - InvalidRecipientPrefixByte, - #[error("failed to decode TransportMessage; too short")] - TransportMessageBytesTooShort, - #[error("failed to decode TransportMessage; invalid nonce")] - InvalidNonce, - #[error("invalid substream ID")] - InvalidSubstreamMessageBytes, - #[error("invalid substream message type byte")] - InvalidSubstreamMessageType, - #[error("substrean with given ID already exists")] - SubstreamIdExists(SubstreamId), - #[error("no substream found for given ID")] - SubstreamIdDoesNotExist(SubstreamId), - #[error("recv error: channel closed")] - OneshotRecvFailure(#[from] tokio::sync::oneshot::error::RecvError), - #[error("recv error: channel closed")] - RecvFailure, - #[error("outbound send error")] - OutboundSendFailure(String), - #[error("inbound send error")] - InboundSendFailure(String), - #[error("failed to send new connection; receiver dropped")] - ConnectionSendFailure, - #[error("failed to send initial TransportEvent::NewAddress")] - SendErrorTransportEvent, - #[error("dial timed out")] - DialTimeout(#[from] tokio::time::error::Elapsed), -} diff --git a/sdk/rust/nym-sdk/examples/libp2p_shared/lib.rs b/sdk/rust/nym-sdk/examples/libp2p_shared/lib.rs deleted file mode 100644 index ba33f1d91f..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_shared/lib.rs +++ /dev/null @@ -1,10 +0,0 @@ -pub(crate) mod connection; -pub mod error; -pub(crate) mod message; -pub(crate) mod mixnet; -pub(crate) mod queue; -pub mod substream; -pub mod transport; - -/// The deafult timeout secs for [`transport::Upgrade`] future. -const DEFAULT_HANDSHAKE_TIMEOUT_SECS: u64 = 5; diff --git a/sdk/rust/nym-sdk/examples/libp2p_shared/message.rs b/sdk/rust/nym-sdk/examples/libp2p_shared/message.rs deleted file mode 100644 index 9dc4c7085b..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_shared/message.rs +++ /dev/null @@ -1,335 +0,0 @@ -use libp2p::core::PeerId; -use nym_sphinx::addressing::clients::Recipient; -use rand::rngs::OsRng; -use rand::RngCore; -use std::fmt::{Debug, Formatter}; - -use super::error::Error; - -const RECIPIENT_LENGTH: usize = Recipient::LEN; -const CONNECTION_ID_LENGTH: usize = 32; -const SUBSTREAM_ID_LENGTH: usize = 32; - -const NONCE_BYTES_LEN: usize = 8; // length of u64 -const MIN_CONNECTION_MESSAGE_LEN: usize = CONNECTION_ID_LENGTH + NONCE_BYTES_LEN; - -/// ConnectionId is a unique, randomly-generated per-connection ID that's used to -/// identify which connection a message belongs to. -#[derive(Clone, Default, Eq, Hash, PartialEq)] -pub(crate) struct ConnectionId([u8; 32]); - -impl ConnectionId { - pub(crate) fn generate() -> Self { - let mut bytes = [0u8; 32]; - OsRng.fill_bytes(&mut bytes); - ConnectionId(bytes) - } - - fn from_bytes(bytes: &[u8]) -> Self { - let mut id = [0u8; 32]; - id[..].copy_from_slice(&bytes[0..CONNECTION_ID_LENGTH]); - ConnectionId(id) - } -} - -impl Debug for ConnectionId { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", hex::encode(self.0)) - } -} - -/// SubstreamId is a unique, randomly-generated per-substream ID that's used to -/// identify which substream a message belongs to. -#[derive(Clone, Default, Eq, Hash, PartialEq)] -pub struct SubstreamId(pub(crate) [u8; 32]); - -impl SubstreamId { - pub(crate) fn generate() -> Self { - let mut bytes = [0u8; 32]; - OsRng.fill_bytes(&mut bytes); - SubstreamId(bytes) - } - - fn from_bytes(bytes: &[u8]) -> Self { - let mut id = [0u8; 32]; - id[..].copy_from_slice(&bytes[0..SUBSTREAM_ID_LENGTH]); - SubstreamId(id) - } -} - -impl Debug for SubstreamId { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", hex::encode(self.0)) - } -} - -#[derive(Debug)] -#[allow(clippy::enum_variant_names)] -pub(crate) enum Message { - ConnectionRequest(ConnectionMessage), - ConnectionResponse(ConnectionMessage), - TransportMessage(TransportMessage), -} - -/// ConnectionMessage is exchanged to open a new connection. -#[derive(Debug)] -pub(crate) struct ConnectionMessage { - pub(crate) peer_id: PeerId, - pub(crate) id: ConnectionId, - /// recipient is the sender's Nym address. - /// only required if this is a ConnectionRequest. - pub(crate) recipient: Option, -} - -/// TransportMessage is sent over a connection after establishment. -#[derive(Debug, Clone)] -pub(crate) struct TransportMessage { - /// increments by 1 for every TransportMessage sent over a connection. - /// required for ordering, since Nym does not guarantee ordering. - /// ConnectionMessages do not need nonces, as we know that they will - /// be the first messages sent over a connection. - /// the first TransportMessage sent over a connection will have nonce 1. - pub(crate) nonce: u64, - pub(crate) message: SubstreamMessage, - pub(crate) id: ConnectionId, -} - -impl Message { - fn try_from_bytes(bytes: Vec) -> Result { - if bytes.len() < 2 { - return Err(Error::InvalidMessageBytes); - } - - Ok(match bytes[0] { - 0 => Message::ConnectionRequest(ConnectionMessage::try_from_bytes(&bytes[1..])?), - 1 => Message::ConnectionResponse(ConnectionMessage::try_from_bytes(&bytes[1..])?), - 2 => Message::TransportMessage(TransportMessage::try_from_bytes(&bytes[1..])?), - _ => return Err(Error::InvalidMessageBytes), - }) - } -} - -impl ConnectionMessage { - fn to_bytes(&self) -> Vec { - let mut bytes = self.id.0.to_vec(); - match self.recipient { - Some(recipient) => { - bytes.push(1u8); - bytes.append(&mut recipient.to_bytes().to_vec()); - } - None => bytes.push(0u8), - } - bytes.append(&mut self.peer_id.to_bytes()); - bytes - } - - fn try_from_bytes(bytes: &[u8]) -> Result { - if bytes.len() < CONNECTION_ID_LENGTH + 1 { - return Err(Error::ConnectionMessageBytesTooShort); - } - - let id = ConnectionId::from_bytes(&bytes[0..CONNECTION_ID_LENGTH]); - let recipient = match bytes[CONNECTION_ID_LENGTH] { - 0u8 => None, - 1u8 => { - if bytes.len() < CONNECTION_ID_LENGTH + 1 + RECIPIENT_LENGTH { - return Err(Error::ConnectionMessageBytesNoRecipient); - } - - let mut recipient_bytes = [0u8; RECIPIENT_LENGTH]; - recipient_bytes[..].copy_from_slice( - &bytes[CONNECTION_ID_LENGTH + 1..CONNECTION_ID_LENGTH + 1 + RECIPIENT_LENGTH], - ); - Some( - Recipient::try_from_bytes(recipient_bytes) - .map_err(Error::InvalidRecipientBytes)?, - ) - } - _ => { - return Err(Error::InvalidRecipientPrefixByte); - } - }; - let peer_id = match recipient { - Some(_) => { - if bytes.len() < CONNECTION_ID_LENGTH + RECIPIENT_LENGTH + 2 { - return Err(Error::ConnectionMessageBytesNoPeerId); - } - PeerId::from_bytes(&bytes[CONNECTION_ID_LENGTH + 1 + RECIPIENT_LENGTH..]) - .map_err(|_| Error::InvalidPeerIdBytes)? - } - None => { - if bytes.len() < CONNECTION_ID_LENGTH + 2 { - return Err(Error::ConnectionMessageBytesNoPeerId); - } - PeerId::from_bytes(&bytes[CONNECTION_ID_LENGTH + 1..]) - .map_err(|_| Error::InvalidPeerIdBytes)? - } - }; - Ok(ConnectionMessage { - peer_id, - recipient, - id, - }) - } -} - -impl TransportMessage { - fn to_bytes(&self) -> Vec { - let mut bytes = self.nonce.to_be_bytes().to_vec(); - bytes.extend_from_slice(self.id.0.as_ref()); - bytes.extend_from_slice(&self.message.to_bytes()); - bytes - } - - fn try_from_bytes(bytes: &[u8]) -> Result { - if bytes.len() < MIN_CONNECTION_MESSAGE_LEN + 1 { - return Err(Error::TransportMessageBytesTooShort); - } - - let nonce = u64::from_be_bytes( - bytes[0..NONCE_BYTES_LEN] - .to_vec() - .try_into() - .map_err(|_| Error::InvalidNonce)?, - ); - let id = ConnectionId::from_bytes(&bytes[NONCE_BYTES_LEN..MIN_CONNECTION_MESSAGE_LEN]); - let message = SubstreamMessage::try_from_bytes(&bytes[MIN_CONNECTION_MESSAGE_LEN..])?; - Ok(TransportMessage { nonce, message, id }) - } -} - -impl Ord for TransportMessage { - fn cmp(&self, other: &Self) -> std::cmp::Ordering { - self.nonce.cmp(&other.nonce) - } -} - -impl std::cmp::PartialOrd for TransportMessage { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} - -impl std::cmp::Eq for TransportMessage {} - -impl std::cmp::PartialEq for TransportMessage { - fn eq(&self, other: &Self) -> bool { - self.nonce == other.nonce - } -} - -#[derive(Debug, Clone)] -pub(crate) enum SubstreamMessageType { - OpenRequest, - OpenResponse, - Close, - Data(Vec), -} - -impl SubstreamMessageType { - fn to_u8(&self) -> u8 { - match self { - SubstreamMessageType::OpenRequest => 0, - SubstreamMessageType::OpenResponse => 1, - SubstreamMessageType::Close => 2, - SubstreamMessageType::Data(_) => 3, - } - } -} - -/// SubstreamMessage is a message sent over a substream. -#[derive(Debug, Clone)] -pub(crate) struct SubstreamMessage { - pub(crate) substream_id: SubstreamId, - pub(crate) message_type: SubstreamMessageType, -} - -impl SubstreamMessage { - pub(crate) fn new_with_data(substream_id: SubstreamId, message: Vec) -> Self { - SubstreamMessage { - substream_id, - message_type: SubstreamMessageType::Data(message), - } - } - - pub(crate) fn new_close(substream_id: SubstreamId) -> Self { - SubstreamMessage { - substream_id, - message_type: SubstreamMessageType::Close, - } - } - - pub(crate) fn to_bytes(&self) -> Vec { - let mut bytes = self.substream_id.0.clone().to_vec(); - bytes.push(self.message_type.to_u8()); - if let SubstreamMessageType::Data(message) = &self.message_type { - bytes.extend_from_slice(message); - } - bytes - } - - pub(crate) fn try_from_bytes(bytes: &[u8]) -> Result { - if bytes.len() < SUBSTREAM_ID_LENGTH + 1 { - return Err(Error::InvalidSubstreamMessageBytes); - } - - let substream_id = SubstreamId::from_bytes(&bytes[0..SUBSTREAM_ID_LENGTH]); - let message_type = match bytes[SUBSTREAM_ID_LENGTH] { - 0 => SubstreamMessageType::OpenRequest, - 1 => SubstreamMessageType::OpenResponse, - 2 => SubstreamMessageType::Close, - 3 => { - if bytes.len() < SUBSTREAM_ID_LENGTH + 2 { - return Err(Error::InvalidSubstreamMessageBytes); - } - SubstreamMessageType::Data(bytes[SUBSTREAM_ID_LENGTH + 1..].to_vec()) - } - _ => return Err(Error::InvalidSubstreamMessageType), - }; - - Ok(SubstreamMessage { - substream_id, - message_type, - }) - } -} - -impl Message { - pub(crate) fn to_bytes(&self) -> Vec { - match self { - Message::ConnectionRequest(msg) => { - let mut bytes = 0_u8.to_be_bytes().to_vec(); - bytes.append(&mut msg.to_bytes()); - bytes - } - Message::ConnectionResponse(msg) => { - let mut bytes = 1_u8.to_be_bytes().to_vec(); - bytes.append(&mut msg.to_bytes()); - bytes - } - Message::TransportMessage(msg) => { - let mut bytes = 2_u8.to_be_bytes().to_vec(); - bytes.append(&mut msg.to_bytes()); - bytes - } - } - } -} - -/// InboundMessage represents an inbound mixnet message. -pub(crate) struct InboundMessage(pub(crate) Message); - -/// OutboundMessage represents an outbound mixnet message. -#[derive(Debug)] -pub(crate) struct OutboundMessage { - pub(crate) message: Message, - pub(crate) recipient: Recipient, -} - -pub(crate) fn parse_message_data(data: &[u8]) -> Result { - if data.len() < 2 { - return Err(Error::InvalidMessageBytes); - } - let msg = Message::try_from_bytes(data.to_vec())?; - Ok(InboundMessage(msg)) -} diff --git a/sdk/rust/nym-sdk/examples/libp2p_shared/mixnet.rs b/sdk/rust/nym-sdk/examples/libp2p_shared/mixnet.rs deleted file mode 100644 index c696bce8fd..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_shared/mixnet.rs +++ /dev/null @@ -1,164 +0,0 @@ -use futures::{pin_mut, select}; -use futures::{FutureExt, StreamExt}; -use log::debug; -use nym_sdk::mixnet::{IncludedSurbs, MixnetClient, MixnetClientSender, MixnetMessageSender}; -use nym_sphinx::addressing::clients::Recipient; -use nym_sphinx::receiver::ReconstructedMessage; -use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}; - -use super::error::Error; -use super::message::*; - -/// initialize_mixnet initializes a read/write connection to a Nym websockets endpoint. -/// It starts a task that listens for inbound messages from the endpoint and writes outbound messages to the endpoint. -pub(crate) async fn initialize_mixnet( - client: MixnetClient, - notify_inbound_tx: Option>, -) -> Result< - ( - Recipient, - UnboundedReceiver, - UnboundedSender, - ), - Error, -> { - let recipient = *client.nym_address(); - - // a channel of inbound messages from the mixnet.. - // the transport reads from (listens) to the inbound_rx. - // TODO: this is probably a DOS vector; we should limit the size of the channel. - let (inbound_tx, inbound_rx) = unbounded_channel::(); - - // a channel of outbound messages to be written to the mixnet. - // the transport writes to outbound_tx. - let (outbound_tx, mut outbound_rx) = unbounded_channel::(); - - let sink = client.split_sender(); - let mut stream = client; - - tokio::task::spawn(async move { - loop { - let t1 = check_inbound(&mut stream, &inbound_tx, ¬ify_inbound_tx).fuse(); - let t2 = check_outbound(&sink, &mut outbound_rx).fuse(); - - pin_mut!(t1, t2); - - select! { - _ = t1 => {}, - _ = t2 => {}, - }; - } - }); - - Ok((recipient, inbound_rx, outbound_tx)) -} - -async fn check_inbound( - client: &mut MixnetClient, - inbound_tx: &UnboundedSender, - notify_inbound_tx: &Option>, -) -> Result<(), Error> { - if let Some(msg) = client.next().await { - if let Some(notify_tx) = notify_inbound_tx { - notify_tx - .send(()) - .map_err(|e| Error::InboundSendFailure(e.to_string()))?; - } - - handle_inbound(msg, inbound_tx).await?; - } - - Err(Error::Unimplemented) -} - -async fn handle_inbound( - msg: ReconstructedMessage, - inbound_tx: &UnboundedSender, -) -> Result<(), Error> { - let data = parse_message_data(&msg.message)?; - inbound_tx - .send(data) - .map_err(|e| Error::InboundSendFailure(e.to_string()))?; - Ok(()) -} - -async fn check_outbound( - mixnet_sender: &MixnetClientSender, - outbound_rx: &mut UnboundedReceiver, -) -> Result<(), Error> { - match outbound_rx.recv().await { - Some(message) => { - write_bytes( - mixnet_sender, - message.recipient, - &message.message.to_bytes(), - ) - .await - } - None => Err(Error::RecvFailure), - } -} - -async fn write_bytes( - mixnet_sender: &MixnetClientSender, - recipient: Recipient, - message: &[u8], -) -> Result<(), Error> { - if let Err(_err) = mixnet_sender - .send_message(recipient, message, IncludedSurbs::ExposeSelfAddress) - .await - { - return Err(Error::Unimplemented); - } - - debug!( - "wrote message to mixnet: recipient: {:?}", - recipient.to_string() - ); - Ok(()) -} - -#[cfg(test)] -mod test { - use super::super::message::{ - self, ConnectionId, Message, SubstreamId, SubstreamMessage, SubstreamMessageType, - TransportMessage, - }; - use super::super::mixnet::initialize_mixnet; - use nym_sdk::mixnet::MixnetClient; - - #[tokio::test] - async fn test_mixnet_poll_inbound_and_outbound() { - let client = MixnetClient::connect_new().await.unwrap(); - let (self_address, mut inbound_rx, outbound_tx) = - initialize_mixnet(client, None).await.unwrap(); - let msg_inner = "hello".as_bytes(); - let substream_id = SubstreamId::generate(); - let msg = Message::TransportMessage(TransportMessage { - nonce: 1, // arbitrary - id: ConnectionId::generate(), - message: SubstreamMessage::new_with_data(substream_id.clone(), msg_inner.to_vec()), - }); - - // send a message to ourselves through the mixnet - let out_msg = message::OutboundMessage { - message: msg, - recipient: self_address, - }; - - outbound_tx.send(out_msg).unwrap(); - - // receive the message from ourselves over the mixnet - let received_msg = inbound_rx.recv().await.unwrap(); - if let Message::TransportMessage(recv_msg) = received_msg.0 { - assert_eq!(substream_id, recv_msg.message.substream_id); - if let SubstreamMessageType::Data(data) = recv_msg.message.message_type { - assert_eq!(msg_inner, data.as_slice()); - } else { - panic!("expected SubstreamMessage::Data") - } - } else { - panic!("expected Message::TransportMessage") - } - } -} diff --git a/sdk/rust/nym-sdk/examples/libp2p_shared/queue.rs b/sdk/rust/nym-sdk/examples/libp2p_shared/queue.rs deleted file mode 100644 index 694b26c65f..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_shared/queue.rs +++ /dev/null @@ -1,138 +0,0 @@ -use log::{debug, warn}; -use std::collections::BTreeSet; - -use super::message::TransportMessage; - -/// MessageQueue is a queue of messages, ordered by nonce, that we've -/// received but are not yet able to process because we're waiting for -/// a message with the next expected nonce first. -/// This is required because Nym does not guarantee any sort of message -/// ordering, only delivery. -/// TODO: is there a DOS vector here where a malicious peer sends us -/// messages only with nonce higher than the next expected nonce? -pub(crate) struct MessageQueue { - /// nonce of the next message we expect to receive on the - /// connection. - /// any messages with a nonce greater than this are pushed into - /// the queue. - /// if we get a message with a nonce equal to this, then we - /// immediately handle it in the transport and increment the nonce. - next_expected_nonce: u64, - - /// the actual queue of messages, ordered by nonce. - /// the head of the queue's nonce is always greater - /// than the next expected nonce. - queue: BTreeSet, -} - -impl MessageQueue { - pub(crate) fn new() -> Self { - MessageQueue { - next_expected_nonce: 0, - queue: BTreeSet::new(), - } - } - - pub(crate) fn print_nonces(&self) { - let nonces = self.queue.iter().map(|msg| msg.nonce).collect::>(); - debug!("MessageQueue: {:?}", nonces); - } - - /// sets the next expected nonce to 1, indicating that we've received - /// a ConnectionRequest or ConnectionResponse. - pub(crate) fn set_connection_message_received(&mut self) { - if self.next_expected_nonce != 0 { - panic!("connection message received twice"); - } - - self.next_expected_nonce = self.next_expected_nonce.wrapping_add(1); - } - - /// tries to push a message into the queue. - /// if the message has the next expected nonce, then the message is returned, - /// and should be processed by the caller. - /// in that case, the internal queue's next expected nonce is incremented. - pub(crate) fn try_push(&mut self, msg: TransportMessage) -> Option { - if msg.nonce == self.next_expected_nonce { - self.next_expected_nonce = self.next_expected_nonce.wrapping_add(1); - Some(msg) - } else { - if msg.nonce < self.next_expected_nonce { - // this shouldn't happen normally, only if the other node - // is not following the protocol - warn!("received a message with a nonce that is too low"); - return None; - } - - if !self.queue.insert(msg) { - // this shouldn't happen normally, only if the other node - // is not following the protocol - warn!("received a message with a duplicate nonce"); - return None; - } - - None - } - } - - pub(crate) fn pop(&mut self) -> Option { - let head = self.queue.first()?; - - if head.nonce == self.next_expected_nonce { - self.next_expected_nonce = self.next_expected_nonce.wrapping_add(1); - Some(self.queue.pop_first().unwrap()) - } else { - None - } - } -} - -#[cfg(test)] -mod test { - use super::super::message::{ConnectionId, SubstreamId, SubstreamMessage}; - - use super::*; - - impl TransportMessage { - fn new(nonce: u64, message: SubstreamMessage, id: ConnectionId) -> Self { - TransportMessage { nonce, message, id } - } - } - - #[test] - fn test_message_queue() { - let mut queue = MessageQueue::new(); - - let test_substream_message = - SubstreamMessage::new_with_data(SubstreamId::generate(), vec![1, 2, 3]); - let connection_id = ConnectionId::generate(); - - let msg1 = TransportMessage::new(1, test_substream_message.clone(), connection_id.clone()); - let msg2 = TransportMessage::new(2, test_substream_message.clone(), connection_id.clone()); - let msg3 = TransportMessage::new(3, test_substream_message.clone(), connection_id.clone()); - - assert_eq!(queue.try_push(msg1.clone()), None); - assert_eq!(queue.try_push(msg3.clone()), None); - assert_eq!(queue.try_push(msg2.clone()), None); - - assert_eq!(queue.pop(), None); - - // set expected nonce to 1 - queue.set_connection_message_received(); - assert_eq!(queue.pop(), Some(msg1)); - - let msg4 = TransportMessage::new(4, test_substream_message.clone(), connection_id.clone()); - assert_eq!(queue.try_push(msg4.clone()), None); - - assert_eq!(queue.pop(), Some(msg2)); - assert_eq!(queue.pop(), Some(msg3)); - assert_eq!(queue.pop(), Some(msg4)); - assert_eq!(queue.pop(), None); - assert_eq!(queue.next_expected_nonce, 5); - - // should just return the message and increment nonce when message nonce = next expected nonce - let msg5 = TransportMessage::new(5, test_substream_message, connection_id); - assert_eq!(queue.try_push(msg5.clone()), Some(msg5)); - assert_eq!(queue.next_expected_nonce, 6); - } -} diff --git a/sdk/rust/nym-sdk/examples/libp2p_shared/substream.rs b/sdk/rust/nym-sdk/examples/libp2p_shared/substream.rs deleted file mode 100644 index e2cd9d7c3e..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_shared/substream.rs +++ /dev/null @@ -1,412 +0,0 @@ -use super::message::{ - ConnectionId, Message, OutboundMessage, SubstreamId, SubstreamMessage, TransportMessage, -}; -use futures::{ - io::{Error as IoError, ErrorKind}, - AsyncRead, AsyncWrite, -}; -use log::debug; -use nym_sphinx::addressing::clients::Recipient; -use parking_lot::Mutex; -use std::{ - pin::Pin, - sync::{ - atomic::{AtomicU64, Ordering}, - Arc, - }, - task::{Context, Poll}, -}; -use tokio::sync::{ - mpsc::{UnboundedReceiver, UnboundedSender}, - oneshot::Receiver, -}; - -#[derive(Debug)] -pub struct Substream { - remote_recipient: Recipient, - connection_id: ConnectionId, - pub(crate) substream_id: SubstreamId, - - /// inbound messages; inbound_tx is in the corresponding Connection - pub(crate) inbound_rx: UnboundedReceiver>, - - /// outbound messages; go directly to the mixnet - outbound_tx: UnboundedSender, - - /// used to signal when the substream is closed - close_rx: Receiver<()>, - closed: Mutex, - - // buffer of data that's been written to the stream, - // but not yet read by the application. - unread_data: Mutex>, - - message_nonce: Arc, -} - -impl Substream { - pub(crate) fn new( - remote_recipient: Recipient, - connection_id: ConnectionId, - substream_id: SubstreamId, - inbound_rx: UnboundedReceiver>, - outbound_tx: UnboundedSender, - close_rx: Receiver<()>, - message_nonce: Arc, - ) -> Self { - Substream { - remote_recipient, - connection_id, - substream_id, - inbound_rx, - outbound_tx, - close_rx, - closed: Mutex::new(false), - unread_data: Mutex::new(vec![]), - message_nonce, - } - } - - fn check_closed(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Result<(), IoError> { - let closed_err = IoError::other("stream closed"); - - // close_rx will return an error if the channel is closed (ie. sender was dropped), - // or if it's empty - let received_closed = self.close_rx.try_recv(); - - let mut closed = self.closed.lock(); - if *closed { - return Err(closed_err); - } - - if received_closed.is_ok() { - *closed = true; - return Err(closed_err); - } - - Ok(()) - } -} - -impl AsyncRead for Substream { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { - let closed_result = self.as_mut().check_closed(cx); - if let Err(e) = closed_result { - return Poll::Ready(Err(e)); - } - - let inbound_rx_data = self.inbound_rx.poll_recv(cx); - - // first, write any previously unread data to the buf - let mut unread_data = self.unread_data.lock(); - let filled_len = if unread_data.len() > 0 { - let unread_len = unread_data.len(); - let buf_len = buf.len(); - let copy_len = std::cmp::min(unread_len, buf_len); - buf[..copy_len].copy_from_slice(&unread_data[..copy_len]); - *unread_data = unread_data[copy_len..].to_vec(); - copy_len - } else { - 0 - }; - - if let Poll::Ready(Some(data)) = inbound_rx_data { - if filled_len == buf.len() { - // we've filled the buffer, so we'll have to save the rest for later - let mut new = vec![]; - new.extend(unread_data.drain(..)); - new.extend(data.iter()); - *unread_data = new; - return Poll::Ready(Ok(filled_len)); - } - - // otherwise, there's still room in the buffer, so we'll copy the rest of the data - let remaining_len = buf.len() - filled_len; - let data_len = data.len(); - - // we have more data than buffer room remaining, save the extra for later - if remaining_len < data_len { - unread_data.extend_from_slice(&data[remaining_len..]); - } - - let copied = std::cmp::min(remaining_len, data_len); - buf[filled_len..filled_len + copied].copy_from_slice(&data[..copied]); - debug!("poll_read copied {} bytes: data {:?}", copied, buf); - return Poll::Ready(Ok(copied)); - } - - if filled_len > 0 { - debug!("poll_read copied {} bytes: data {:?}", filled_len, buf); - return Poll::Ready(Ok(filled_len)); - } - - Poll::Pending - } -} - -impl AsyncWrite for Substream { - fn poll_write( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - if let Err(e) = self.as_mut().check_closed(cx) { - return Poll::Ready(Err(e)); - } - - let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst); - - self.outbound_tx - .send(OutboundMessage { - recipient: self.remote_recipient, - message: Message::TransportMessage(TransportMessage { - nonce, - id: self.connection_id.clone(), - message: SubstreamMessage::new_with_data( - self.substream_id.clone(), - buf.to_vec(), - ), - }), - }) - .map_err(|e| IoError::other(format!("poll_write outbound_tx error: {}", e)))?; - - Poll::Ready(Ok(buf.len())) - } - - fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - if let Err(e) = self.check_closed(cx) { - return Poll::Ready(Err(e)); - } - - Poll::Ready(Ok(())) - } - - fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst); - - let mut closed = self.closed.lock(); - if *closed { - return Poll::Ready(Err(IoError::other("stream closed"))); - } - - *closed = true; - - // send a close message to the mixnet - self.outbound_tx - .send(OutboundMessage { - recipient: self.remote_recipient, - message: Message::TransportMessage(TransportMessage { - nonce, - id: self.connection_id.clone(), - message: SubstreamMessage::new_close(self.substream_id.clone()), - }), - }) - .map_err(|e| IoError::other(format!("poll_close outbound_rx error: {}", e)))?; - - Poll::Ready(Ok(())) - } -} - -#[cfg(test)] -mod test { - use super::super::message::{ - ConnectionId, Message, SubstreamId, SubstreamMessage, TransportMessage, - }; - use super::super::mixnet::initialize_mixnet; - use super::Substream; - use futures::{AsyncReadExt, AsyncWriteExt}; - use nym_sdk::mixnet::MixnetClient; - use nym_sphinx::addressing::clients::Recipient; - use std::sync::atomic::AtomicU64; - use std::sync::Arc; - - #[tokio::test] - async fn test_substream_poll_read_unread_data() { - let (outbound_tx, _) = tokio::sync::mpsc::unbounded_channel(); - let connection_id = ConnectionId::generate(); - let substream_id = SubstreamId::generate(); - - let (inbound_tx, inbound_rx) = tokio::sync::mpsc::unbounded_channel(); - let (_, close_rx) = tokio::sync::oneshot::channel(); - - let mut substream = Substream::new( - Recipient::try_from_base58_string("D1rrpsysCGCYXy9saP8y3kmNpGtJZUXN9SvFoUcqAsM9.9Ssso1ea5NfkbMASdiseDSjTN1fSWda5SgEVjdSN4CvV@GJqd3ZxpXWSNxTfx7B1pPtswpetH4LnJdFeLeuY5KUuN").unwrap(), - connection_id, - substream_id, - inbound_rx, - outbound_tx, - close_rx, - Arc::new(AtomicU64::new(1)), - ); - - // test writing and reading w/ same length data - let data = b"hello".to_vec(); - inbound_tx.send(data.clone()).unwrap(); - let mut buf = [0u8; 5]; - let read_len = substream.read(&mut buf).await.unwrap(); - assert_eq!(read_len, data.len()); - assert_eq!(buf.to_vec(), data); - - // test writing data longer than read buffer - let data = b"nootwashere".to_vec(); - inbound_tx.send(data.clone()).unwrap(); - - let mut buf = [0u8; 4]; - let read_len = substream.read(&mut buf).await.unwrap(); - assert_eq!(read_len, buf.len()); - assert_eq!(buf.to_vec(), b"noot".to_vec()); - - let mut buf = [0u8; 7]; - let read_len = substream.read(&mut buf).await.unwrap(); - assert_eq!(read_len, buf.len()); - assert_eq!(buf.to_vec(), b"washere".to_vec()); - - // test read buffer larger than written data - let data = b"nootwashere".to_vec(); - inbound_tx.send(data.clone()).unwrap(); - let mut buf = [0u8; 16]; - let read_len = substream.read(&mut buf).await.unwrap(); - assert_eq!(read_len, data.len()); - assert_eq!(buf[..data.len()], data); - assert_eq!(buf[data.len()..].to_vec(), vec![0u8; 16 - data.len()]); - - // test writing data longer than read buffer multiple times - let data = b"nootwashere".to_vec(); - inbound_tx.send(data.clone()).unwrap(); - - let mut buf = [0u8; 4]; - let read_len = substream.read(&mut buf).await.unwrap(); - assert_eq!(read_len, buf.len()); - assert_eq!(buf.to_vec(), b"noot".to_vec()); - - let data = b"asdf".to_vec(); - inbound_tx.send(data.clone()).unwrap(); - - let mut buf = [0u8; 4]; - let read_len = substream.read(&mut buf).await.unwrap(); - assert_eq!(read_len, buf.len()); - assert_eq!(buf.to_vec(), b"wash".to_vec()); - - let mut buf = [0u8; 8]; - let read_len = substream.read(&mut buf).await.unwrap(); - assert_eq!(read_len, 7); - assert_eq!(buf[..7], b"ereasdf".to_vec()); - } - - #[tokio::test] - async fn test_substream_read_write() { - let client = MixnetClient::connect_new().await.unwrap(); - let (self_address, mut mixnet_inbound_rx, outbound_tx) = - initialize_mixnet(client, None).await.unwrap(); - - const MSG_INNER: &[u8] = "hello".as_bytes(); - let connection_id = ConnectionId::generate(); - let substream_id = SubstreamId::generate(); - - let (inbound_tx, inbound_rx) = tokio::sync::mpsc::unbounded_channel(); - let (_, close_rx) = tokio::sync::oneshot::channel(); - - let mut substream = Substream::new( - self_address, - connection_id, - substream_id, - inbound_rx, - outbound_tx, - close_rx, - Arc::new(AtomicU64::new(1)), - ); - - // send message to ourselves over the mixnet - substream.write_all(MSG_INNER).await.unwrap(); - - // receive full message over the mixnet - let recv_msg = mixnet_inbound_rx.recv().await.unwrap(); - match recv_msg.0 { - Message::TransportMessage(TransportMessage { - nonce, - id: _, - message: - SubstreamMessage { - substream_id: _, - message_type: msg, - }, - }) => { - assert_eq!(nonce, 1); - match msg { - super::super::message::SubstreamMessageType::Data(data) => { - assert_eq!(data, MSG_INNER); - // send message to substream inbound channel - inbound_tx.send(data).unwrap(); - } - _ => panic!("unexpected message type"), - } - } - _ => panic!("unexpected message"), - } - - // read message from substream - let mut buf = [0u8; MSG_INNER.len()]; - substream.read_exact(&mut buf).await.unwrap(); - assert_eq!(buf, MSG_INNER); - - // close substream - substream.close().await.unwrap(); - - // try to read/write to closed substream; should error - substream.write_all(MSG_INNER).await.unwrap_err(); - substream.read_exact(&mut buf).await.unwrap_err(); - - // assert a close message was sent over the mixnet - let recv_msg = mixnet_inbound_rx.recv().await.unwrap(); - match recv_msg.0 { - Message::TransportMessage(TransportMessage { - nonce: _, - id: _, - message: - SubstreamMessage { - substream_id: _, - message_type: msg, - }, - }) => match msg { - super::super::message::SubstreamMessageType::Close => {} - _ => panic!("unexpected message type"), - }, - _ => panic!("unexpected message: {:?}", recv_msg.0), - } - } - - #[tokio::test] - async fn test_substream_recv_close() { - let client = MixnetClient::connect_new().await.unwrap(); - let (self_address, _, outbound_tx) = initialize_mixnet(client, None).await.unwrap(); - - const MSG_INNER: &[u8] = "hello".as_bytes(); - let connection_id = ConnectionId::generate(); - let substream_id = SubstreamId::generate(); - - let (_, inbound_rx) = tokio::sync::mpsc::unbounded_channel(); - let (close_tx, close_rx) = tokio::sync::oneshot::channel(); - - let mut substream = Substream::new( - self_address, - connection_id, - substream_id, - inbound_rx, - outbound_tx, - close_rx, - Arc::new(AtomicU64::new(1)), - ); - - // close substream - close_tx.send(()).unwrap(); - - // try to read/write to closed substream; should error - substream.write_all(MSG_INNER).await.unwrap_err(); - let mut buf = [0u8; MSG_INNER.len()]; - substream.read_exact(&mut buf).await.unwrap_err(); - } -} diff --git a/sdk/rust/nym-sdk/examples/libp2p_shared/transport.rs b/sdk/rust/nym-sdk/examples/libp2p_shared/transport.rs deleted file mode 100644 index 46a81a8577..0000000000 --- a/sdk/rust/nym-sdk/examples/libp2p_shared/transport.rs +++ /dev/null @@ -1,898 +0,0 @@ -use futures::prelude::*; -use libp2p::core::{ - identity::Keypair, - multiaddr::{Multiaddr, Protocol}, - transport::{ListenerId, TransportError, TransportEvent}, - PeerId, Transport, -}; -use log::debug; -use nym_sdk::mixnet::MixnetClient; -use nym_sphinx::addressing::clients::Recipient; -use std::{ - collections::HashMap, - pin::Pin, - str::FromStr, - task::{Context, Poll, Waker}, -}; -use tokio::{ - sync::{ - mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}, - oneshot, - }, - time::{timeout, Duration}, -}; -use tokio_stream::wrappers::UnboundedReceiverStream; - -use super::connection::{Connection, PendingConnection}; -use super::error::Error; -use super::message::{ - ConnectionId, ConnectionMessage, InboundMessage, Message, OutboundMessage, SubstreamMessage, - TransportMessage, -}; -use super::mixnet::initialize_mixnet; -use super::queue::MessageQueue; -use super::DEFAULT_HANDSHAKE_TIMEOUT_SECS; - -/// InboundTransportEvent represents an inbound event from the mixnet. -pub enum InboundTransportEvent { - ConnectionRequest(Upgrade), - ConnectionResponse, - TransportMessage, -} - -/// NymTransport implements the Transport trait using the Nym mixnet. -pub struct NymTransport { - /// our Nym address - self_address: Recipient, - pub(crate) listen_addr: Multiaddr, - pub(crate) listener_id: ListenerId, - - /// our libp2p keypair; currently not really used - keypair: Keypair, - - /// established connections -> channel which sends messages received from - /// the mixnet to the corresponding Connection - connections: HashMap>, - - /// outbound pending dials - pending_dials: HashMap, - - /// connection message queues - message_queues: HashMap, - - /// inbound mixnet messages - inbound_stream: UnboundedReceiverStream, - - /// outbound mixnet messages - outbound_tx: UnboundedSender, - - /// inbound messages for Transport.poll() - poll_rx: UnboundedReceiver>, - - /// outbound messages to Transport.poll() - poll_tx: UnboundedSender>, - - waker: Option, - - /// Timeout for the [`Upgrade`] future. - handshake_timeout: Duration, -} - -impl NymTransport { - /// New transport. - #[allow(unused)] - pub async fn new(client: MixnetClient, keypair: Keypair) -> Result { - Self::new_maybe_with_notify_inbound(client, keypair, None, None).await - } - - /// New transport with a timeout. - #[allow(dead_code)] - pub async fn new_with_timeout( - client: MixnetClient, - keypair: Keypair, - timeout: Duration, - ) -> Result { - Self::new_maybe_with_notify_inbound(client, keypair, None, Some(timeout)).await - } - - /// Add timeout to transport and return self. - #[allow(dead_code)] - pub fn with_timeout(mut self, timeout: Duration) -> Self { - self.handshake_timeout = timeout; - self - } - - async fn new_maybe_with_notify_inbound( - client: MixnetClient, - keypair: Keypair, - notify_inbound_tx: Option>, - timeout: Option, - ) -> Result { - let (self_address, inbound_rx, outbound_tx) = - initialize_mixnet(client, notify_inbound_tx).await?; - let listen_addr = nym_address_to_multiaddress(self_address)?; - let listener_id = ListenerId::new(); - - let (poll_tx, poll_rx) = unbounded_channel::>(); - - poll_tx - .send(TransportEvent::NewAddress { - listener_id, - listen_addr: listen_addr.clone(), - }) - .map_err(|_| Error::SendErrorTransportEvent)?; - - let inbound_stream = UnboundedReceiverStream::new(inbound_rx); - let handshake_timeout = - timeout.unwrap_or_else(|| Duration::from_secs(DEFAULT_HANDSHAKE_TIMEOUT_SECS)); - - Ok(Self { - self_address, - listen_addr, - listener_id, - keypair, - connections: HashMap::new(), - pending_dials: HashMap::new(), - message_queues: HashMap::new(), - inbound_stream, - outbound_tx, - poll_rx, - poll_tx, - waker: None, - handshake_timeout, - }) - } - - pub(crate) fn peer_id(&self) -> PeerId { - PeerId::from_public_key(&self.keypair.public()) - } - - fn handle_message_queue_on_connection_initiation( - &mut self, - id: &ConnectionId, - ) -> Result<(), Error> { - debug!("handle_message_queue_on_connection_initiation"); - let Some(inbound_tx) = self.connections.get(id) else { - // this should not happen - return Err(Error::NoConnectionForTransportMessage); - }; - - match self.message_queues.get_mut(id) { - Some(queue) => { - // update expected nonce - queue.set_connection_message_received(); - - // push pending inbound some messages in this case - while let Some(msg) = queue.pop() { - debug!( - "popped queued message with nonce {} for connection", - msg.nonce - ); - inbound_tx - .send(msg.message.clone()) - .map_err(|e| Error::InboundSendFailure(e.to_string()))?; - } - } - None => { - // no queue exists for this connection, create one - let queue = MessageQueue::new(); - self.message_queues.insert(id.clone(), queue); - let queue = self.message_queues.get_mut(id).unwrap(); - queue.set_connection_message_received(); - } - }; - - debug!("returning from handle_message_queue_on_connection_initiation"); - Ok(()) - } - - // handle_connection_response resolves the pending connection corresponding to the response - // (if there is one) into a Connection. - fn handle_connection_response(&mut self, msg: &ConnectionMessage) -> Result<(), Error> { - if self.connections.contains_key(&msg.id) { - return Err(Error::ConnectionAlreadyEstablished); - } - - if let Some(pending_conn) = self.pending_dials.remove(&msg.id) { - // resolve connection and put into pending_conn channel - let (conn, conn_tx) = self.create_connection_types( - msg.peer_id, - pending_conn.remote_recipient, - msg.id.clone(), - ); - - self.connections.insert(msg.id.clone(), conn_tx); - self.handle_message_queue_on_connection_initiation(&msg.id)?; - - pending_conn - .connection_tx - .send(conn) - .map_err(|_| Error::ConnectionSendFailure)?; - - if let Some(waker) = self.waker.take() { - waker.wake(); - } - - Ok(()) - } else { - Err(Error::NoConnectionForResponse) - } - } - - /// handle_connection_request handles an incoming connection request, sends back a - /// connection response, and finally completes the upgrade into a Connection. - fn handle_connection_request(&mut self, msg: &ConnectionMessage) -> Result { - if msg.recipient.is_none() { - return Err(Error::NoneRecipientInConnectionRequest); - } - - // ensure we don't already have a conn with the same id - if self.connections.contains_key(&msg.id) { - return Err(Error::ConnectionIDExists); - } - - let (conn, conn_tx) = - self.create_connection_types(msg.peer_id, msg.recipient.unwrap(), msg.id.clone()); - self.connections.insert(msg.id.clone(), conn_tx); - self.handle_message_queue_on_connection_initiation(&msg.id)?; - - let resp = ConnectionMessage { - peer_id: self.peer_id(), - recipient: None, - id: msg.id.clone(), - }; - - self.outbound_tx - .send(OutboundMessage { - message: Message::ConnectionResponse(resp), - recipient: msg.recipient.unwrap(), - }) - .map_err(|e| Error::OutboundSendFailure(e.to_string()))?; - - if let Some(waker) = self.waker.take() { - waker.wake(); - } - - Ok(conn) - } - - fn handle_transport_message(&mut self, msg: TransportMessage) -> Result<(), Error> { - let queue = match self.message_queues.get_mut(&msg.id) { - Some(queue) => queue, - None => { - // no queue exists for this connection, create one - let queue = MessageQueue::new(); - self.message_queues.insert(msg.id.clone(), queue); - self.message_queues.get_mut(&msg.id).unwrap() - } - }; - - queue.print_nonces(); - - let nonce = msg.nonce; - let Some(msg) = queue.try_push(msg) else { - // don't push the message yet, it's been queued - debug!("message with nonce {} queued for connection", nonce); - return Ok(()); - }; - - let Some(inbound_tx) = self.connections.get(&msg.id) else { - return Err(Error::NoConnectionForTransportMessage); - }; - - // send original message - debug!( - "sending original message with nonce {} for connection", - nonce - ); - inbound_tx - .send(msg.message.clone()) - .map_err(|e| Error::InboundSendFailure(e.to_string()))?; - - // try to pop queued messages and send them on inbound channel - while let Some(msg) = queue.pop() { - debug!( - "popped queued message with nonce {} for connection", - msg.nonce - ); - inbound_tx - .send(msg.message.clone()) - .map_err(|e| Error::InboundSendFailure(e.to_string()))?; - } - - if let Some(waker) = self.waker.clone().take() { - waker.wake(); - } - - Ok(()) - } - - fn create_connection_types( - &self, - remote_peer_id: PeerId, - recipient: Recipient, - id: ConnectionId, - ) -> (Connection, UnboundedSender) { - let (inbound_tx, inbound_rx) = unbounded_channel::(); - - // representation of a connection; this contains channels for applications to read/write to. - let conn = Connection::new( - remote_peer_id, - recipient, - id, - inbound_rx, - self.outbound_tx.clone(), - ); - - // inbound_tx is what we write to when receiving messages on the mixnet, - (conn, inbound_tx) - } - - /// handle_inbound handles an inbound message from the mixnet, received via self.inbound_stream. - fn handle_inbound(&mut self, msg: Message) -> Result { - match msg { - Message::ConnectionRequest(inner) => { - debug!("got inbound connection request {:?}", inner); - match self.handle_connection_request(&inner) { - Ok(conn) => { - let (connection_tx, connection_rx) = - oneshot::channel::<(PeerId, Connection)>(); - let upgrade = Upgrade::new(connection_rx); - connection_tx - .send((inner.peer_id, conn)) - .map_err(|_| Error::ConnectionSendFailure)?; - Ok(InboundTransportEvent::ConnectionRequest(upgrade)) - } - Err(e) => Err(e), - } - } - Message::ConnectionResponse(msg) => { - debug!("got inbound connection response {:?}", msg); - self.handle_connection_response(&msg) - .map(|_| InboundTransportEvent::ConnectionResponse) - } - Message::TransportMessage(msg) => { - debug!("got inbound TransportMessage: {:?}", msg); - self.handle_transport_message(msg) - .map(|_| InboundTransportEvent::TransportMessage) - } - } - } -} - -/// Upgrade represents a transport listener upgrade. -/// Note: we immediately upgrade a connection request to a connection, -/// so this only contains a channel for receiving that connection. -pub struct Upgrade { - connection_tx: oneshot::Receiver<(PeerId, Connection)>, -} - -impl Upgrade { - fn new(connection_tx: oneshot::Receiver<(PeerId, Connection)>) -> Upgrade { - Upgrade { connection_tx } - } -} - -impl Future for Upgrade { - type Output = Result<(PeerId, Connection), Error>; - - // poll checks if the upgrade has turned into a connection yet - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - self.connection_tx - .poll_unpin(cx) - .map_err(|_| Error::RecvFailure) - } -} - -impl Transport for NymTransport { - type Output = (PeerId, Connection); - type Error = Error; - type ListenerUpgrade = Upgrade; - type Dial = Pin> + Send>>; - - fn listen_on(&mut self, _: Multiaddr) -> Result> { - // we should only allow listening on the multiaddress containing our Nym address - Ok(self.listener_id) - } - - fn remove_listener(&mut self, id: ListenerId) -> bool { - if self.listener_id != id { - return false; - } - - // TODO: close channels? - self.poll_tx - .send(TransportEvent::ListenerClosed { - listener_id: id, - reason: Ok(()), - }) - .expect("failed to send listener closed event"); - true - } - - fn dial(&mut self, addr: Multiaddr) -> Result> { - debug!("dialing {}", addr); - - let id = ConnectionId::generate(); - - // create remote recipient address - let recipient = multiaddress_to_nym_address(addr).map_err(TransportError::Other)?; - - // create pending conn structs and store - let (connection_tx, connection_rx) = oneshot::channel::(); - - let inner_pending_conn = PendingConnection::new(recipient, connection_tx); - self.pending_dials.insert(id.clone(), inner_pending_conn); - - // put ConnectionRequest message into outbound message channel - let msg = ConnectionMessage { - peer_id: self.peer_id(), - recipient: Some(self.self_address), - id, - }; - - let outbound_tx = self.outbound_tx.clone(); - - let mut waker = self.waker.clone(); - let handshake_timeout = self.handshake_timeout; - Ok(async move { - outbound_tx - .send(OutboundMessage { - message: Message::ConnectionRequest(msg), - recipient, - }) - .map_err(|e| Error::OutboundSendFailure(e.to_string()))?; - - debug!("sent outbound ConnectionRequest"); - if let Some(waker) = waker.take() { - waker.wake(); - }; - - let conn = timeout(handshake_timeout, connection_rx).await??; - Ok((conn.peer_id, conn)) - } - .boxed()) - } - - // dial_as_listener currently just calls self.dial(). - fn dial_as_listener( - &mut self, - addr: Multiaddr, - ) -> Result> { - self.dial(addr) - } - - fn poll( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll> { - // new addresses + listener close events - if let Poll::Ready(Some(res)) = self.poll_rx.recv().boxed().poll_unpin(cx) { - return Poll::Ready(res); - } - - // check for and handle inbound messages - while let Poll::Ready(Some(msg)) = self.inbound_stream.poll_next_unpin(cx) { - match self.handle_inbound(msg.0) { - Ok(event) => match event { - InboundTransportEvent::ConnectionRequest(upgrade) => { - debug!("InboundTransportEvent::ConnectionRequest"); - return Poll::Ready(TransportEvent::Incoming { - listener_id: self.listener_id, - upgrade, - local_addr: self.listen_addr.clone(), - send_back_addr: self.listen_addr.clone(), - }); - } - InboundTransportEvent::ConnectionResponse => { - debug!("InboundTransportEvent::ConnectionResponse"); - } - InboundTransportEvent::TransportMessage => { - debug!("InboundTransportEvent::TransportMessage"); - } - }, - Err(e) => { - return Poll::Ready(TransportEvent::ListenerError { - listener_id: self.listener_id, - error: e, - }); - } - }; - } - - self.waker = Some(cx.waker().clone()); - Poll::Pending - } - - fn address_translation(&self, _listen: &Multiaddr, _observed: &Multiaddr) -> Option { - None - } -} - -fn nym_address_to_multiaddress(addr: Recipient) -> Result { - Multiaddr::from_str(&format!("/nym/{}", addr)).map_err(Error::FailedToFormatMultiaddr) -} - -fn multiaddress_to_nym_address(multiaddr: Multiaddr) -> Result { - let mut multiaddr = multiaddr; - match multiaddr.pop().unwrap() { - Protocol::Nym(addr) => Recipient::from_str(&addr).map_err(Error::InvalidRecipientBytes), - _ => Err(Error::InvalidProtocolForMultiaddr), - } -} - -#[cfg(test)] -mod test { - use super::super::connection::Connection; - use super::super::error::Error; - use super::super::message::{ - Message, OutboundMessage, SubstreamId, SubstreamMessage, SubstreamMessageType, - TransportMessage, - }; - use super::super::substream::Substream; - use super::{nym_address_to_multiaddress, NymTransport}; - use futures::{future::poll_fn, AsyncReadExt, AsyncWriteExt, FutureExt}; - use libp2p::core::{ - identity::Keypair, - transport::{Transport, TransportEvent}, - Multiaddr, StreamMuxer, - }; - use log::info; - use nym_bin_common::logging::setup_tracing_logger; - use nym_sdk::mixnet::MixnetClient; - use std::{pin::Pin, str::FromStr, sync::atomic::Ordering}; - use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}; - - impl Connection { - fn write(&self, msg: SubstreamMessage) -> Result<(), Error> { - let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst); - self.mixnet_outbound_tx - .send(OutboundMessage { - recipient: self.remote_recipient, - message: Message::TransportMessage(TransportMessage { - nonce, - id: self.id.clone(), - message: msg, - }), - }) - .map_err(|e| Error::OutboundSendFailure(e.to_string()))?; - Ok(()) - } - } - - impl NymTransport { - async fn new_with_notify_inbound( - client: MixnetClient, - notify_inbound_tx: UnboundedSender<()>, - ) -> Result { - let local_key = Keypair::generate_ed25519(); - Self::new_maybe_with_notify_inbound(client, local_key, Some(notify_inbound_tx), None) - .await - } - } - - #[tokio::test] - async fn test_transport_connection() { - setup_tracing_logger(); - - let client = MixnetClient::connect_new().await.unwrap(); - let (dialer_notify_inbound_tx, mut dialer_notify_inbound_rx) = unbounded_channel(); - let mut dialer_transport = - NymTransport::new_with_notify_inbound(client, dialer_notify_inbound_tx) - .await - .unwrap(); - - let client2 = MixnetClient::connect_new().await.unwrap(); - let (listener_notify_inbound_tx, mut listener_notify_inbound_rx) = unbounded_channel(); - let mut listener_transport = - NymTransport::new_with_notify_inbound(client2, listener_notify_inbound_tx) - .await - .unwrap(); - let listener_multiaddr = - nym_address_to_multiaddress(listener_transport.self_address).unwrap(); - assert_new_address_event(Pin::new(&mut dialer_transport)).await; - assert_new_address_event(Pin::new(&mut listener_transport)).await; - - // dial the remote peer - let mut dial = dialer_transport.dial(listener_multiaddr).unwrap(); - - // poll the dial to send the connection request message - assert!(poll_fn(|cx| Pin::new(&mut dial).as_mut().poll_unpin(cx)) - .now_or_never() - .is_none()); - listener_notify_inbound_rx.recv().await.unwrap(); - - // should receive the connection request from the mixnet and send the connection response - let res = poll_fn(|cx| Pin::new(&mut listener_transport).as_mut().poll(cx)).await; - let mut upgrade = match res { - TransportEvent::Incoming { - listener_id, - upgrade, - local_addr, - send_back_addr, - } => { - assert_eq!(listener_id, listener_transport.listener_id); - assert_eq!(local_addr, listener_transport.listen_addr); - assert_eq!(send_back_addr, listener_transport.listen_addr); - upgrade - } - _ => panic!("expected TransportEvent::Incoming, got {:?}", res), - }; - dialer_notify_inbound_rx.recv().await.unwrap(); - - // should receive the connection response from the mixnet - assert!( - poll_fn(|cx| Pin::new(&mut dialer_transport).as_mut().poll(cx)) - .now_or_never() - .is_none() - ); - info!("waiting for connections..."); - - // should be able to resolve the connections now - let (_, mut listener_conn) = poll_fn(|cx| Pin::new(&mut upgrade).as_mut().poll_unpin(cx)) - .now_or_never() - .expect("the upgrade should be ready") - .expect("the upgrade should not error"); - let (_, mut dialer_conn) = poll_fn(|cx| Pin::new(&mut dial).as_mut().poll_unpin(cx)) - .now_or_never() - .expect("the upgrade should be ready") - .expect("the upgrade should not error"); - info!("connections established"); - - // write messages from the dialer to the listener and vice versa - send_and_receive_over_conns( - b"hello".to_vec(), - &mut dialer_conn, - &mut listener_conn, - Pin::new(&mut listener_transport), - &mut listener_notify_inbound_rx, - ) - .await; - send_and_receive_over_conns( - b"hi".to_vec(), - &mut dialer_conn, - &mut listener_conn, - Pin::new(&mut listener_transport), - &mut listener_notify_inbound_rx, - ) - .await; - send_and_receive_over_conns( - b"world".to_vec(), - &mut listener_conn, - &mut dialer_conn, - Pin::new(&mut dialer_transport), - &mut dialer_notify_inbound_rx, - ) - .await; - } - - async fn assert_new_address_event(mut transport: Pin<&mut NymTransport>) { - match poll_fn(|cx| transport.as_mut().poll(cx)).await { - TransportEvent::NewAddress { - listener_id, - listen_addr, - } => { - assert_eq!(listener_id, transport.listener_id); - assert_eq!(listen_addr, transport.listen_addr); - } - _ => panic!("expected TransportEvent::NewAddress"), - } - } - - async fn send_and_receive_over_conns( - msg: Vec, - conn1: &mut Connection, - conn2: &mut Connection, - mut transport2: Pin<&mut NymTransport>, - notify_inbound_rx: &mut UnboundedReceiver<()>, - ) { - // send message over conn1 to conn2 - let substream_id = SubstreamId::generate(); - conn1 - .write(SubstreamMessage::new_with_data( - substream_id.clone(), - msg.clone(), - )) - .unwrap(); - notify_inbound_rx.recv().await.unwrap(); - - // poll transport2 to push message from transport to connection - assert!(poll_fn(|cx| transport2.as_mut().poll(cx)) - .now_or_never() - .is_none()); - let substream_msg = conn2.inbound_rx.recv().await.unwrap(); - if let SubstreamMessageType::Data(data) = substream_msg.message_type { - assert_eq!(data, msg); - } else { - panic!("expected data message"); - } - } - - #[tokio::test] - async fn test_transport_substream() { - let client = MixnetClient::connect_new().await.unwrap(); - - let (dialer_notify_inbound_tx, mut dialer_notify_inbound_rx) = unbounded_channel(); - let mut dialer_transport = - NymTransport::new_with_notify_inbound(client, dialer_notify_inbound_tx) - .await - .unwrap(); - - let client2 = MixnetClient::connect_new().await.unwrap(); - - let (listener_notify_inbound_tx, mut listener_notify_inbound_rx) = unbounded_channel(); - let mut listener_transport = - NymTransport::new_with_notify_inbound(client2, listener_notify_inbound_tx) - .await - .unwrap(); - let listener_multiaddr = - nym_address_to_multiaddress(listener_transport.self_address).unwrap(); - assert_new_address_event(Pin::new(&mut dialer_transport)).await; - assert_new_address_event(Pin::new(&mut listener_transport)).await; - - // dial the remote peer - let mut dial = dialer_transport.dial(listener_multiaddr).unwrap(); - - // poll the dial to send the connection request message - assert!(poll_fn(|cx| Pin::new(&mut dial).as_mut().poll_unpin(cx)) - .now_or_never() - .is_none()); - listener_notify_inbound_rx.recv().await.unwrap(); - - // should receive the connection request from the mixnet and send the connection response - let res = poll_fn(|cx| Pin::new(&mut listener_transport).as_mut().poll(cx)).await; - let mut upgrade = match res { - TransportEvent::Incoming { - listener_id, - upgrade, - local_addr, - send_back_addr, - } => { - assert_eq!(listener_id, listener_transport.listener_id); - assert_eq!(local_addr, listener_transport.listen_addr); - assert_eq!(send_back_addr, listener_transport.listen_addr); - upgrade - } - _ => panic!("expected TransportEvent::Incoming, got {:?}", res), - }; - dialer_notify_inbound_rx.recv().await.unwrap(); - - // should receive the connection response from the mixnet - assert!( - poll_fn(|cx| Pin::new(&mut dialer_transport).as_mut().poll(cx)) - .now_or_never() - .is_none() - ); - info!("waiting for connections..."); - - // should be able to resolve the connections now - let (_, mut listener_conn) = poll_fn(|cx| Pin::new(&mut upgrade).as_mut().poll_unpin(cx)) - .now_or_never() - .unwrap() - .unwrap(); - let (_, mut dialer_conn) = poll_fn(|cx| Pin::new(&mut dial).as_mut().poll_unpin(cx)) - .now_or_never() - .unwrap() - .unwrap(); - info!("connections established"); - - // initiate a new substream from the dialer - let mut dialer_substream = - poll_fn(|cx| Pin::new(&mut dialer_conn).as_mut().poll_outbound(cx)) - .await - .unwrap(); - listener_notify_inbound_rx.recv().await.unwrap(); - - // accept the substream on the listener - assert!( - poll_fn(|cx| Pin::new(&mut listener_transport).as_mut().poll(cx)) - .now_or_never() - .is_none() - ); - poll_fn(|cx| Pin::new(&mut listener_conn).as_mut().poll(cx)).now_or_never(); - - // poll recipient's poll_inbound to receive the substream; sends a response to the sender - let mut listener_substream = - poll_fn(|cx| Pin::new(&mut listener_conn).as_mut().poll_inbound(cx)) - .now_or_never() - .unwrap() - .unwrap(); - info!("got listener substream"); - dialer_notify_inbound_rx.recv().await.unwrap(); - - // poll sender to finalize the substream - assert!( - poll_fn(|cx| Pin::new(&mut dialer_transport).as_mut().poll(cx)) - .now_or_never() - .is_none() - ); - poll_fn(|cx| Pin::new(&mut dialer_conn).as_mut().poll(cx)).now_or_never(); - info!("got dialer substream"); - - // write message from dialer to listener - send_and_receive_substream_message( - b"hello world".to_vec(), - Pin::new(&mut dialer_substream), - Pin::new(&mut listener_substream), - Pin::new(&mut listener_transport), - Pin::new(&mut listener_conn), - &mut listener_notify_inbound_rx, - ) - .await; - - // write message from listener to dialer - send_and_receive_substream_message( - b"hello back".to_vec(), - Pin::new(&mut listener_substream), - Pin::new(&mut dialer_substream), - Pin::new(&mut dialer_transport), - Pin::new(&mut dialer_conn), - &mut dialer_notify_inbound_rx, - ) - .await; - - // close the substream from the dialer side - info!("closing dialer substream"); - dialer_substream.close().await.unwrap(); - listener_notify_inbound_rx.recv().await.unwrap(); - info!("dialer substream closed"); - - // assert we can't read or write to either substream - dialer_substream.write_all(b"hello").await.unwrap_err(); - // poll listener transport and conn to receive the substream close message - poll_fn(|cx| Pin::new(&mut listener_transport).as_mut().poll(cx)).now_or_never(); - poll_fn(|cx| Pin::new(&mut listener_conn).as_mut().poll(cx)).now_or_never(); - listener_substream.write_all(b"hello").await.unwrap_err(); - let mut buf = vec![0u8; 5]; - dialer_substream.read(&mut buf).await.unwrap_err(); - listener_substream.read(&mut buf).await.unwrap_err(); - dialer_substream.close().await.unwrap_err(); - listener_substream.close().await.unwrap_err(); - } - - async fn send_and_receive_substream_message( - data: Vec, - mut sender_substream: Pin<&mut Substream>, - mut recipient_substream: Pin<&mut Substream>, - mut recipient_transport: Pin<&mut NymTransport>, - mut recipient_conn: Pin<&mut Connection>, - recipient_notify_inbound_rx: &mut UnboundedReceiver<()>, - ) { - // write message - sender_substream.write_all(&data).await.unwrap(); - recipient_notify_inbound_rx.recv().await.unwrap(); - - // poll recipient for message - poll_fn(|cx| recipient_transport.as_mut().poll(cx)).now_or_never(); - poll_fn(|cx| recipient_conn.as_mut().poll(cx)).now_or_never(); - let mut buf = vec![0u8; data.len()]; - let n = recipient_substream.read(&mut buf).await.unwrap(); - assert_eq!(n, data.len()); - assert_eq!(buf, data[..]); - } - - #[tokio::test] - async fn test_transport_timeout() { - let client = MixnetClient::connect_new().await.unwrap(); - - let (dialer_notify_inbound_tx, _) = unbounded_channel(); - let mut dialer_transport = - NymTransport::new_with_notify_inbound(client, dialer_notify_inbound_tx) - .await - .unwrap() - .with_timeout(std::time::Duration::from_millis(100)); - - // mock a transport that will never resolve the connection. - let empty_addr = Multiaddr::from_str( - "/nym/Hmer6Ndt3PV13YW53HM8ri4NvqqtfDQUQBhzvKqb1dag.2g478dyxtrQXGWc1Mk2VEqdPcWXpz7EhAcjhdAJtVZdA@AnnYnEtBjB2a5sHmeRCnBq43qxyHDf95Bqd7cwQyKNLR" - ) - .expect("unable to parse multiaddress"); - - let dial = dialer_transport.dial(empty_addr).unwrap(); - assert!(dial - .await - .expect_err("should have timed out") - .to_string() - .contains("dial timed out")); - } -}